[아파치 카프카 입문과 활용] 5. Apache kafka 파이프라인 실습

www.youtube.com/watch?v=OX6s1-bYBI4&list=PL3Re5Ri5rZmkY46j6WcJXQYRlDRZSUQ1j&index=20

 

Kafka 활용 실습

서버 메트릭 수집 및 적재 파이프라인 만들기

• 서버 메트릭(cpu, ram, network)을 수집
• 이슈 확인 및 모니터링을 위해 사용
• 그라파나를 활용하면 그래프 시각화 및 slack alert 가능
  그라파나(Grafana)란? - 로그 시각화 도구로서 키바나와
  비슷한 역할. 다양한 db source 지원

 

Metric collect pipeline

요구 사항

• CPU, RAM 정보 수집
  👉 Telegraf agent 사용 : https://github.com/influxdata/telegraf
  Telegraf : influxdata에서 opensource로 제공하는 plugin 기반의 metric수집 server agent
  간단한 configuration으로 다양한 plugin을 붙 일 수 있음

[agent]
  interval = "10s"

[[outputs.kafka]]
  brokers = ["{aws ec2 public ip}:9092"]
  ## Kafka topic for producer messages
  topic = "my-computer-metric"

[[inputs.cpu]]
  percpu = true
  totalcpu = true
  fielddrop = ["time_*"]

[[inputs.mem]]

• 파일단위로 저장
  • write와 함께 flush
  • 파일 포맷은 CSV(Comma Separater values)

$ cat 2020-06-19.csv
mem,host=SKP1003855MA0001.local used=19689496576i,inactive=14526488576i, ...

• 카프카 관련 정보
  • 토픽명 : my-computer-metric
  • 파티션 개수 : 5개
  • 컨슈머 개수 : 5개(multiple thread)
  • 키 활용 여부 : 사용하지 않음

 

 

PRACTICE - Metric collect pipeline

1. Topic 생성

• 토픽명 : my-computer-metric
• 파티션 개수 : 5개
• 레플리케이션 : 1개(브로커가 1대이므로)

$ ./kafka-topics.sh --create --bootstrap-server {aws ec2 public ip}:9092 --replication-factor 1 -- partitions 5 --topic my-computer-metric

2. Telegraf 설치

👉 Homebrew(package 관리도구)를 사용하여 install
👉 https://brew.sh/index_ko

$ brew update
$ brew install telegraf
$ brew info telegraf
telegraf: stable 1.16.3 (bottled), HEAD
Server-level metric gathering agent for InfluxDB
https://influxdata.com
/usr/local/Cellar/telegraf/1.16.3 (9 files, 100.2MB) *

telegraf 설치위치로 이동

$ cd /usr/local/Cellar/telegraf/1.16.3/bin

3. Telegraf config 저장

https://github.com/AndersonChoi/tacademy-kafka/blob/master/telegraf.conf

telegraf.conf 파일 생성

[agent]
  interval = "10s"

[[outputs.kafka]]
  brokers = ["13.125.138.28:9092"]
  ## Kafka topic for producer messages
  topic = "my-computer-metric"

[[inputs.cpu]]
  percpu = true
  totalcpu = true
  fielddrop = ["time_*"]

[[inputs.mem]]

4. Telegraf 실행

telegraf 에이전트 실행 10초 마다 수행

./telegraf -config telegraf.conf
or
telegraf -config /usr/local/etc/telegraf.conf

$ ./telegraf --config telegraf.conf
2020-06-09T10:51:25Z I! Starting Telegraf 1.16.3
2020-06-09T10:51:25Z I! Loaded inputs: cpu mem
2020-06-09T10:51:25Z I! Loaded aggregators:
2020-06-09T10:51:25Z I! Loaded processors:
2020-06-09T10:51:25Z I! Loaded outputs: kafka
2020-06-09T10:51:25Z I! Tags enabled: host=SKP1003855MA0001.local

5. kafka-console-consumer 데이터 확인

카프카 활용 실습 토픽 데이터 확인

$ ./kafka-console-consumer.sh --bootstrap-server {aws ec2 public ip}:9092 --topic my-computer-metric --from-beginning

cpu,cpu=cpu2,host=SKP1003855MA0001.local
usage_system=3.2,usage_idle=90.8,usage_guest=0,usage_steal=0,usage_guest_nice=0,usage_user=6,usa ge_nice=0,usage_iowait=0,usage_irq=0,usage_softirq=0 1591699990000000000

cpu,cpu=cpu1,host=SKP1003855MA0001.local
usage_user=0.1,usage_system=0.1,usage_idle=99.8,usage_irq=0,usage_guest=0,usage_guest_nice=0,usa ge_nice=0,usage_iowait=0,usage_softirq=0,usage_steal=0 1591699990000000000

6. Kafka consumer application 개발

• File write는 Java에서 제공하는 FileWriter 클래스사용

File file = new File("kafka.csv");
FileWriter fw = new FileWriter(file, true);
fw.write("hello kafka");
fw.close();

7. multiple thread 사용, group.id 지정

FileWriter fw = new FileWriter(file, true);
StringBuilder fileWriteBuffer = new StringBuilder();
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
  fileWriteBuffer.append(record.value());
}
fw.write(fileWriteBuffer.toString());
consumer.commitSync();
fw.close();

8. 파일 적재 확인

kafka-consumer-save-metric

• CONSUMER_COUNT = 5 로 변경
  private static int CONSUMER_COUNT = 5;

프로젝트 내의 csv 파일에 지속적으로 기록되는 것 확인

$ tail -f *.csv

cpu,cpu=cpu2,host=SKP1003855MA0001.local
usage_guest_nice=0,usage_user=21.1,usage_system=32.1,usage_idle=46.8,usage_nice=0,usage_iowait=0 ,usage_irq=0,usage_guest=0,usage_softirq=0,usage_steal=0 1591759660000000000

cpu,cpu=cpu1,host=SKP1003855MA0001.local
usage_softirq=0,usage_steal=0,usage_guest_nice=0,usage_system=1,usage_idle=98.5,usage_nice=0,usa ge_iowait=0,usage_irq=0,usage_guest=0,usage_user=0.5 1591759660000000000