Apache Kafka [1] - 개요 및 아키텍처

카프카는 이벤트스크리밍(Event Streaming)기반 메세징시스템(Messaging System)으로 실시간 데이터를 저장, 조작, 다른목적지로 배치작업 등을 하기에 용이하다. 

메세징시스템이란?

메세징시스템은 한 어플리케이션에서 다른 어플리케이션으로 데이터를 전송하는 역할을 한다. 분산메세징시스템은 메세지 큐를 기반으로 어플리케이션과 메세지 큐 사이에 비동기적(Asynchronously)으로 작동한다.

보통 2개의 타입의 메세징패턴이 제공되는데 포인트 투 포인트(point-point)와 게시구독(pub-sub)시스템이다.

포인트 투 포인트 메세징시스템(point to point messaging system)

대표적인 예로 주문처리시스템을 생각하면 된다. 한 명 이상의 소비자가 큐에 있는 메세지를 소비할 수 있지만 특정 메세지는 한 명의 소비자만 소비할 수 있다.

게시-구독 메세징시스템(publish-subscribe messaging system)

포인트 투 포인트 시스템과 달리 소비자들은 하나 이상의 큐에있는 데이터를 소비할 수 있다. 게시-구독 시스템에선 생산자를 Publisher라 하고, 메세지 소비자를 Subscriber라고 한다. 간단한 예로 TV채널을 들 수 있는데, 누구나 자신이 가입한 채널을 시청할 수 있듯이 Subscriber들은 연결되기만 하면 누구든 큐에서 데이터를 가져다 올 수 있다.

카프카도 분산된 게시-구독 메세징시스템이며, 대량의 데이터를 처리함과 동시에 데이터손실을 막기위해 데이터들은 클러스터 내에 복제된다. 카프카는 Zookeeper서비스 위에서 작동하며 아파치 스톰, 스파크 등과 연동하여 실시간 스트리밍분석 또한 용이하다.

카프카의 장점(Benefits)

• 성능(Performance) - 카프카는 고성능 TCP네트워크 프로토콜을 통해 통신하는 분산형 시스템이다. 이는 모든 쓰기가 OS(RAM) 캐시로 수행하기 때문에 가능한데, 때문에 초당 200만번의 쓰기 작업을 수행 할 수 있다. 

• 신뢰성(Reliability) - 대기시간이 짧은 메세지 전달을 지원하며, 분산/분할/복제를 통해 시스템 장애 발생 시 결함되지 않는 안정성을 보장한다. 

• 확장성(Scalability) - 프로듀서와 컨슈머의 분리로 Hive, Hbase, Python 등과 같이 다양한 어플리케이션과 사용하기 쉬우며 다운타임(down time)없이 쉽게 확장 가능하다.

• 내구성(Durability) - 카프카는 분산커밋로그(Distributed commit log)를 사용하여, 중간에 데이터를 잃어도 재생할 수 있어 내구성이 뛰어나다. 

주요 개념(main concepts)

- 카프카 아키텍처(kafka architecture)

• Topics - 특정 범주에 속하는 메세지 스트림. 토픽을 통해 Producers와 Consumers가 데이터를 전송한다.

• Partition - Topic을 이루는 단위. 토픽은 수많은 파티션으로 구성된다.

• Partition offset - 파티션을 구분하는 시퀀스ID

• Replicas of partition - 파티션 백업을 위한 복제본. 데이터를 읽거나 쓰지 않으며 단지 데이터 손실이 있을 경우 사용된다.

• Brokers - 데이터를 관리하는 시스템. 각 브로커는 주제당 0개 이상의 파티션을 가질 수 있다.

• Kafka Cluster - 카프카 브로커 1개 이상이 있는 것을 클러스터라고 부른다. 클러스터는 다운타임(down time)없이 확장할 수 있으며 메세지 데이터의 지속성과 복제를 관리하는데 사용된다.

• Producers - 프로듀서는 하나이상의 카프카 토픽(Topic)을 제공한다. 프로듀서는 토픽의 파티션에 데이터를 전송하며 메세지를 퍼블리싱할 때마다 브로커가 파시션의 마지막 세그먼트 파일에 메세지를 첨부한다. 

• Consumers - 컨슈머는 하나 이상의 토픽에 연결되어 있고 브로커로부터 데이터를 읽는다.

• Leader - 리더는 주어진 파티션에 대한 모든 읽기,쓰기를 담당하는 노드다. 모든 파티션에는 리더 역할을 하는 하나의 서버가 있다. 

• Follower - 리더의 지시를 따르는 노드. 리더가 결함시 팔로워 중 한명이 자동으로 새로운 리더가 된다. 팔로워는 메세지를 끌어들이고 자신의 데이터 저장소를 업데이트 한다.

• Consumer Group - 컨슈머의 집합을 구성하는 단위. 컨슈머 그룹 안의 컨슈머 수만큼 파티션의 데이터를 분산처리한다.

- 프로듀서와 컨슈머 Read/Write Work Flow

Apaceh JMeter - 설치 및 실행

[ 개요 ]

제이미터는 웹프로그램을 테스트하기 위한 JAVA애플리케이션이다.

현재는 다양한 테스트가 가능하도록 기능이 확장되었다.

• Web - HTTP, HTTPS (Java, NodeJS, PHP, ASP.NET, …)
• SOAP / REST Webservices
• FTP
• Database via JDBC
• LDAP
• Message-oriented middleware (MOM) via JMS
• Mail - SMTP(S), POP3(S) and IMAP(S)
• Native commands or shell scripts
• TCP
• Java Objects

[ 설치 ]

해당 URL에서 https://jmeter.apache.org/download_jmeter.cgi Binaries의 zip파일을 클릭한다. 

설치하려는 환경에 자바가 설치되어있어야한다

[ 실행 ]

압축을 풀고 bin 폴더에 있는 jmeterw.cmd파일을 실행

[ 예제 ]

HTTP Request를 이용해 해당 input태그에 값을 POST로 넘기고 확인해보는 예제이다.

1. 좌측 약병이모티콘 우클릭 -> add -> Threads(Users) -> Thread Group

2. Thread Group우클릭 -> add -> Sampler -> Http Request

3. Thread Group우클릭 -> add -> Listener -> View Results Tree / View Results in Table

* Http Request로 파라미터를 넣고 실행하고 View Results Tree / View Results in Table로 해당 결과를 확인한다.

4. HTML CODE

<form action="upload_file.php" method="post" enctype="multipart/form-data">

<label for="file">Filename:</label>

<input type="file" name="file1" id="file1"><br>

<input type="text" name="position1"><br>

<input type="text" name="action1"><br>

<input type="text" name="imageCode1"><br>

<input type="text" name="width1"><br>

<input type="text" name="height1"><br>

.....

<input type="submit" name="submit" value="Submit">

</form>

5. 파라미터 값 입력

테스트 데이터들을 아래그림과 같이 Input 태그의 Name에 맞춰서 넣는다.

6. 업로드 테스트할 파일 파라미터 입력

7. 실행 후 확인

상단의 메뉴표시바에서 실행버튼을 클릭 후 View Results Tree / View Results in Table 통해 확인.

Sampler result, Request, Response data탭을 활용하여 리턴된 값, Http 상태등을 체크

이상으로 간단히 HTTP Request를 통해 해당 웹페이지에 값을 POST로 보내고 결과를 확인해 보았다. 파라미터 값 뿐만 아니라 csv, json파일로도 테스트가 가능하다. 

자세한 기능은 https://jmeter.apache.org/ 을 참조한다.

InfluxDB [1] - 설치 및 실행

 

[ 개요 ]

InfluxDB는 쓰기쿼리를 빠르게 처리하도록 설계된 시계열 데이터베이스(TSDB)이다. 기존 RDBS와 다르게 시간을 기준으로 데이터를 입력하므로 금융데이터, 로그데이터 등 시간에 따른 데이터를 분석하기에 적합하다. InfluxDB와 함께 Telegraf(모니터링 및 수집), Grafana(시계열데이터 시각화) 등과 함께 사용된다.

[ 설치 ]

InfluxDB 1.8

- Red Hat & CentOS

cat <<EOF | sudo tee /etc/yum.repos.d/influxdb.repo
[influxdb]
name = InfluxDB Repository - RHEL \$releasever
baseurl = https://repos.influxdata.com/rhel/\$releasever/\$basearch/stable
enabled = 1
gpgcheck = 1
gpgkey = https://repos.influxdata.com/influxdb.key
EOF

sudo yum install influxdb
sudo service influxdb start

- Ubuntu

wget -qO- https://repos.influxdata.com/influxdb.key | sudo apt-key add -
source /etc/lsb-release
echo "deb https://repos.influxdata.com/${DISTRIB_ID,,} ${DISTRIB_CODENAME} stable" | sudo tee /etc/apt/sources.list.d/influxdb.list

sudo apt-get update && sudo apt-get install influxdb
sudo service influxdb start

[ 설정파일 ]

/etc/influxdb/influxdb.conf

[ 실행 ]

$ influx -precision rfc3339

InfluxDB는 기본적으로 포트 8086에서 실행된다. -precision옵션은 타임스탬프의 포맷을 지정한다. 위 예에서 rfc3339는 (YYYY-MM-DDH:MM:SS.nnnnnnnZ)형식의 타임스탬프를 반환하도록 하는 옵션이다.

[ 예제 ]

구성요소

• Points : 시간값과 measurement, key-value tag와 field로 구성된 단위
• Measurement : RDBS에서 Table과 같은 개념
• Field : 인덱스되지 않은 column
• Tags : 인덱스된 column

- DB생성

> CREATE DATABASE mydb

> SHOW DATABASES

> USE mydb

- 데이터쓰기

<measurement>[,<tag-key>=<tag-value>...] <field-key>=<field-value>[, <field2-key>=<field2-value>...] [unix-nano-timestamp]

> INSERT cpu,host=serverA,region=us_west value=0.64  //cpu measurement가 없다면 자동생성

> SHOW MEASUREMENTS  //cpu measurement가 검색됨

- 결과

> SELECT "host", "region", "value" FROM "cpu"

name: cpu

time                           host    region  value

----                           ----    ------  -----

2020-09-13T12:21:37.928191574Z serverA us_west 0.64

> SELECT * FROM /.*/ LIMIT1

> SELECT * FROM "cpu"

> SELECT * FROM "cpu" WHERE "value" > 0.5

Ref : https://docs.influxdata.com/influxdb/v1.8/introduction/get-started/

Apache Sqoop [1] - 설치[ Install ]

[ 개요 ]

전 프로젝트에서 mysql데이터를 hive table에 주기적으로 input하는 기능이 필요했다. 이런 상황에서 Data-pip-line을 구성하기 좋고 적용하기도 쉬운 Sqoop에 대해 알아보자.  

스쿱(Sqoop)은 관계형데이터베이스(Ex. Mysql,MariaDB)와 아파치 하둡간의 데이터 인아웃을 위한 CLI(Command Line Interface) 어플리케이션이다. RDB에서 데이터를 가져와 맵리듀스작업을 할 수 있고 그 반대도 가능하다. 장애컨트롤 뿐만 아니라 병렬처리까지 가능한 것이 장점이다.

[ 아키텍처 ]

중간에 위치한 Sqoop Tool은 Java로 프로그래밍되어 SQL to Hadoop, Hadoop to SQL 변환을 자유롭게 해준다.

Import : RDBMS에서 개별테이블을 가져와 한 행씩 HDFS 레코드로 처리된다. 레코드는 텍스트데이터, Avro, 이진데이터로 병렬로 저장된다. 

Export : HDFS에서 RDBMS 행데이터가 있는 파일집합을 만들어 내보낸다. 

[ 설치 ]

일단 Java와 Hadoop이 설치되어 있어야 한다.

설치할 경로에 스쿱 tar.gz파일 다운받는다.

$ cd /usr/local
$ wget http://mirror.navercorp.com/apache/sqoop/1.4.7/sqoop-1.4.7.bin__hadoop-2.6.0.tar.gz
$ tar -xvf sqoop-1.4.7.tar.gz

스쿱관련 설정을 해준다.

$ vi /etc/profile
//아래 코드 추가
export SQOOP_HOME=/usr/local/sqoop-1.4.7.bin__hadoop-2.6.0
export PATH=$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$SQOOP_HOME/bin:$PATH
$ source /etc/profile

$ cd $SQOOP_HOME/conf
$ mv sqoop-env-template.sh sqoop-env.sh

$ vi sqoop-env.sh
//주석을 제거하고 아래 코드를 추가한다.
export HADOOP_COMMON_HOME=/usr/local/hadoop/hadoop-3.2.1
export HADOOP_MAPRED_HOME=/usr/local/hadoop/hadoop-3.2.1

위에서 Sqoop Tool은 Java기반으로 동작하기 때문에 mysql-connector-java-5.1.30.tar.gz을 다운받아서 Sqoop경로에 복사한다.

$ wget https://downloads.mysql.com/archives/get/p/3/file/mysql-connector-java-5.1.30.tar.gz
$ cd /usr/local/mysql-connector-java-5.1.30
$ cp mysql-connector-java-5.1.30-bin.jar /usr/local/sqoop-1.4.7.bin__hadoop-2.6.0/lib/

설치 확인

$ sqoop-version

Apache HBase [1] - 개요 및 아키텍처

1.Hbase의 등장

하둡은 MapReduce를 이용하여 대용량 데이터를 저장하고 처리하는데는 탁월하다. 그러나 데이터를 순차적방식으로 액세스하여, 간단한 작업을 할 때에도 해당 데이터를 찾을 때 까지 데이터 집합 전체를 검색해야 한다.

이러한 불편함을 개선하기 위해 Random Access Database를 개발하였는데 HBase, Cassandra, couchDB, Dynamo, MongoDB와 같은 데이터베이스이다.

2.HBase란

[ 컬럼지향데이터베이스 ]

HBase는 HDFS(Hadoop Distributed File System)위에 구축된 분산형 컬럼지향 데이터베이스이다. ROW기반의 RDBMS보다 데이터를 통계화 하는데 이점이 있다. 

[ 이미지참조 http://www.dbguide.net/ ]

위의 예와 같이 column1의 평균을 구하려면 RDBMS에서는 Group By로 묶은 후 평균을 구하는 "SELECT AVG(column1) FROM DB1 GROUP BY column1" 와 같이 그룹핑 단계를 거쳐야 하지만 컬럼지향데이터베이스는 그럴 필요가 없어 매우 빠른 속도로 결과값을 얻을 수 있다.

RDBMS는 Online Transaction Process(OLTP)에 적합하여 비즈니스 트랜잭션 처리에 많이 사용한다. 쇼핑몰 구매시스템이 그 예다. 반면 컬럼기반 HBase는 Online Analytical Processing(OLAP)에 적합하여 데이터 분석, 통계 모델링 구축에 많이 사용된다.

[ 랜덤 액세스 ]

기존 HDFS와 가장 큰 차이점은 순차적 데이터액세스만 제공했던 HDFS와는 다르게, HBase는 내부적 Hash테이블을 사용한 실시간 읽기/쓰기 랜덤 액세스를 제공한다. 따라서 보다 빠른 데이터처리를 할 수 있다.

3. HBase 저장 매커니즘

HBase의 테이블 스키마는 키-값 쌍으로 된 컬럼패밀리(column family)로 정의된다. 컬럼패밀리는 수 많은 컬럼을 가질 수 있으며, 각 컬럼은 타임스탬프 값을 가진다.

4. HBase의 특징

• HBase는 행기준으로 확장이 가능하다. ( 스키마 정보만 있으면 클러스터A에는 1-100행, 클러스터B에는 100-300행을 따로 저장하여 확장기능을 가능케한다. )
• HBase는 Failure support 기능이 있다.
• 일관된 읽기과 쓰기를 제공한다.
• Hadoop및 관련 애플리케이션(Phoenix, Hive 등)과 사용이 용이하다.
• 클라이언트를 위한 쉬운 JAVA API를 제공한다.
• 클러스터 간에 데이터 복제 기능을 제공하여 데이터의 정밀성과 정합성을 보장한다.

5. HBase 아키텍처

HBase는 컬럼패밀리에 의해 수직으로 분리된 스토어(MemStore+H file)로 나뉘며, region server에 의해 각 스토어가 HDFS에 저장된다. 지역서버(region server)는 추가되거나 제거할 수 있다.

Hbase 구성요소

• 클라이언트 라이브러리(Client Library)
• 마스터서버(Master Server)
• 지역서버(Region Server)

* HDFS와 Zookeeper와 함께 연동

[ 마스터서버 ]

- Zookeeper를 통해 각 지역서버를 할당하고 작업을 지시

- 서버간 로드밸런싱

- 스키마 및 테이블, 컬럼패밀리 생성/삭제와 같은 메타데이터 작업처리

[ 지역서버 ]

- 클라이언트와 통신하여 데이터관련 작업처리(읽기/쓰기)

- 데이터 및 서버의 Data Size 임계값을 결정

- HDFS 위에서 동작한다.

WAL(Write Ahead Log) : 새로운 데이터를 저장하기 위해 쓰이는데 중간에 데이터를 잃을 경우 데이터 복구 역할도 한다.

Block Cache : 자주 읽는 데이터를 메모리에 저장

Memstore : Disk에 새로운 데이터를 Write한다. key-value기반의 Cache메모리로 속도가 빠르며 region의 하나의 컬럼패밀리마다 memstore를 갖는다.

Hfile : key-value로 저장된 데이터파일

[ 주키퍼 ]

- 마스터서버가 사용가능한 서버와 통신(서버장애확인,사용가능서버검색)하기 위해 사용 

 

[SQL] 0으로 나눌 때 exception

SELECT ISNULL(col1, 0 ) FROM table1
SELECT COALESCE(col1, 0 ) FROM table1

Comparing COALESCE() and ISNULL():
1. The ISNULL function and the COALESCE expression have a similar purpose but can behave differently.

2. Because ISNULL is a function, it is evaluated only once. As described above, the input values for the COALESCE expression can be evaluated multiple times.

3. Data type determination of the resulting expression is different. ISNULL uses the data type of the first parameter, COALESCE follows the CASE expression rules and returns the data type of value with the highest precedence.

4. The NULLability of the result expression is different for ISNULL and COALESCE. The ISNULL return value is always considered NOT NULLable (assuming the return value is a non-nullable one) whereas COALESCE with non-null parameters is considered to be NULL. So the expressions ISNULL(NULL, 1) and COALESCE(NULL, 1) although equivalent have different nullability values. This makes a difference if you are using these expressions in computed columns, creating key constraints or making the return value of a scalar UDF deterministic so that it can be indexed as shown in the following example.

[SQL]resultSet 함수

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.ResultSetMetaData;

String sql = "select * from board where id=?";

PreparedStatement pstmt = conn.prepareStatement(sql);

ResultSet rs = pstmt.executeQuery();

ResultSetMetaData rsmd = rs.getMetaData();

int columnCnt = rsmd.getColumnCount(); //컬럼의 수

 

if(rs.next()){

  for(int i=1 ; i<=columnCnt ; i++){

                                    // 컬럼명                                   //데이터

System.out.println(rsmd.getColumnName(i)+","+rs.getString(rsmd.getColumnName(i)));  

  }

}