Kafka 설치

Kafka Installation

# Java 1.8.0 Upper install needed
$ yum install java

$ curl -o kafka_2.13-2.8.0.tgz https://mirror.navercorp.com/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz

$ tar -xzf kafka_2.13-2.8.0.tgz
$ cd kafka_2.13-2.8.0

# Startup
$ bin/zookeeper-server-start.sh config/zookeeper.properties &
$ bin/kafka-server-start.sh config/server.properties &

# Topic Create ( Default )
$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

Kafka Cluster Installation

Cluster 구성시 Kafka가 구동 중이라면 중지한다.

클러스터로 구성된 두 시스템 그룹 간 네트워크 혹은 서버 장애로 인해서 일시적으로 단절 현상이 발생하여 나타나는 문제를 스플릿 브레인(Split Brain) 현상이라고 한다.

이러한 문제가 발생하게 될 경우 클러스터 상의 모든 노드가 각자 자신이 Primary로 인식하게 되어 중복으로 서비스가 실행하게 된다. 이로 인해 데이터 동기화 문제로 인해서 추후 문제가 해결되었을 때 정상적인 운영이 불가능하게 되므로 클러스터를 구성하는 의미 자체가 사라지게 된다.

따라서 2대의 구성은 추천되고 있지 않으며, 2대 이상의 구성에서는 오버헤드의 감소를 위해서 홀수의 구성이 추천되고 있다.

# Zookeeper config setting - Common ( Not Standalone - Quorum Mode )
vi config/zookeeper.properties

initLimit=5 # 팔로워가 리더와 초기에 연결하는 시간에 대한 타임아웃
syncLimit=2 # 팔로워가 리더와 동기화 하는데에 대한 타임아웃. 즉 이 틱 시간안에 팔로워가 리더와 동기화가 되지 않는다면 제거 된다.

server.1=Server1:2888:3888
server.2=Server2:2888:3888
server.3=Server3:2888:3888


# /tmp/zookeeper 디렉토리가 없다면 생성하고 myid 파일을 생성하여 각각 서버의 고유 ID값을 부여해야 한다.


# Server1
$ mkdir /tmp/zookeeper $ echo 1 > /tmp/zookeeper/myid
$ vi server.properties

broker.id=1
advertised.listeners=PLAINTEXT://Server1:9092
zookeeper.connect=Server1:2181, Server2:2181, Server3:2181


# Server2
$ mkdir /tmp/zookeeper $ echo 2 > /tmp/zookeeper/myid
$ vi server.properties

broker.id=2
advertised.listeners=PLAINTEXT://Server2:9092
zookeeper.connect=Server1:2181, Server2:2181, Server3:2181


# Server3
$ mkdir /tmp/zookeeper $ echo 3 > /tmp/zookeeper/myid
$ vi server.properties

broker.id=3
advertised.listeners=PLAINTEXT://Server3:9092
zookeeper.connect=Server1:2181, Server2:2181, Server3:2181


# Startup - Common
$ bin/zookeeper-server-start.sh config/zookeeper.properties &
$ bin/kafka-server-start.sh config/server.propertie &

# Kafka Replication Topic 구성 Example
$ ./kafka-topics.sh --zookeeper Server1:2181,Server2:2181,Server3:2181 --create --topic test-topic --partitions 3 --replication-factor 3

Ksql Installation

# Download the public key and follow prompts to import it.
$ curl -sq http://ksqldb-packages.s3.amazonaws.com/archive/0.18/archive.key | gpg --import

# Download the archive and its signature
$ curl http://ksqldb-packages.s3.amazonaws.com/archive/0.18/confluent-ksqldb-0.18.0.tar.gz --output confluent-ksqldb-0.18.0.tar.gz
$ curl http://ksqldb-packages.s3.amazonaws.com/archive/0.18/confluent-ksqldb-0.18.0.tar.gz.asc --output confluent-ksqldb-0.18.0.tar.gz.asc

# Validate the signature
$ gpg --verify confluent-ksqldb-0.18.0.tar.gz.asc confluent-ksqldb-0.18.0.tar.gz

# Extract the tarball to the directory of your choice
$ tar -xf confluent-ksqldb-0.18.0.tar.gz -C /opt/confluent


$vi /opt/confluent/confluent-ksqldb-0.18.0/etc/ksqldb/ksql-server.properties

#------ Kafka -------

# The set of Kafka brokers to bootstrap Kafka cluster information from:
bootstrap.servers=localhost:9092

# Enable snappy compression for the Kafka producers
compression.type=snappy


# Startup KSQL
$ sudo /opt/confluent/confluent-ksqldb-0.18.0/bin/ksql-server-start /opt/confluent/confluent-ksqldb-0.18.0/etc/ksqldb/ksql-server.properties


# Startup Cli
$ /opt/confluent/confluent-ksqldb-0.18.0/bin/ksql http://0.0.0.0:8088

# Example Stream Create
CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE)
  WITH (kafka_topic='locations', value_format='json', partitions=1);

# Watching riderLocations Streams
SELECT * FROM riderLocations
  WHERE GEO_DISTANCE(latitude, longitude, 37.4133, -122.1162) <= 5 EMIT CHANGES;

# Insert Streams Event using Another Cli
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('c2309eec', 37.7877, -122.4205);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('18f4ea86', 37.3903, -122.0643);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4ab5cbad', 37.3952, -122.0813);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('8b6eae59', 37.3944, -122.0813);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4a7c7b41', 37.4049, -122.0822);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4ddad000', 37.7857, -122.4011);

Kafka Go Client Installation ( with C++ Lib ) , Provider "confluentinc"

wrapping C++ Library ( using CGO )

$sudo yum groupinstall "Development Tools"

$sudo yum install librdkafka-devel 

#if not exist librdkafka-devel
$sudo rpm --import https://packages.confluent.io/rpm/6.2/archive.key

$vi /etc/yum.repo.d
[Confluent.dist]
name=Confluent repository (dist)
baseurl=https://packages.confluent.io/rpm/6.2/$releasever
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/6.2/archive.key
enabled=1

[Confluent]
name=Confluent repository
baseurl=https://packages.confluent.io/rpm/6.2
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/6.2/archive.key
enabled=1


$go get gopkg.in/confluentinc/confluent-kafka-go.v1/kafka

$go get gopkg.in/confluentinc/confluent-kafka-go.v1/examples/go-kafkacat

Sarama Kafka Client Library Installation , Provider "Shopify"

go get github.com/Shopify/sarama