<div id="readability-page-1" class="page"><div id="kakaoContent">

 <div>
 <a href="/m/category/%EC%8A%A4%ED%84%B0%EB%94%94">스터디</a><h2>Filebeat-Kafka-Nifi로 데이터 스트림 5분 안에 구성해보기</h2>
 <cite>
 chancethecoder
 </cite>
 </div>
 <time>2019. 12. 22. 15:04</time>
 
 
 <a href="#comment" onclick="Utils.smoothScroll(this);return false">
 
 </a>
 
 
 
 
 
 
 
 

 
 <div>
 이번 포스트에서는 Filebeat, Kafka, Nifi를 docker-compose로 구성하고, 데이터 스트림을 생성하는 과정을 진행해보겠습니다. 튜토리얼 방식으로, 개인 로컬 맥에서 간단하게 따라서 진행해주시면 됩니다.
<figure width="100%" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/cu4bId/btqALQBeyXp/3pskhMqG8qTYZilf1W0hmK/img.png" width="100%" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"></figure>
데이터는 Filebeat -&gt; Kafka -&gt; Nifi -&gt; Kafka 순서로 흘러가는 구조입니다. 데이터 소스로 Apache 웹 로그를 사용해 실제 환경과 같이 일정 속도로 흘러가도록 구성하고, 실시간으로 데이터 스트림이 동작하는 모습을 확인해보겠습니다.
Tested on:
<ul>
<li>MacOS v10.14.6</li>
<li>docker v19.03</li>
<li>docker-compose v1.24</li>
</ul>
<h4 data-ke-size="size20">구성하기</h4>
<ol>
<li>
적당한 workspace로 이동한 뒤, Filebeat에 제공하기 위한 데이터 소스로 weblogs.log 파일을 생성합니다. 일단 빈 파일로 생성하고, 추후에 데이터를 추가할 예정입니다.
<pre><code>touch weblogs.log</code></pre>
</li>
<li>
filebeat.yml 파일을 생성하고 아래 내용을 추가합니다.
<pre><code>filebeat:
 inputs:
 - type: log
 paths:
 - /data/weblogs.log
 config:
 modules:
 path: ${path.config}/modules.d/*.yml
 reload.enabled: false

output:
 kafka:
 hosts: ['kafka:29092']
 topic: 'filebeats-weblogs'
 partition:
 round_robin:
 reachable_only: false
 required_acks: 1
 compression: gzip
 max_message_bytes: 1000000</code></pre>
</li>
<li>
docker-compose.yml 파일을 생성하고 아래 내용을 추가합니다.
<pre><code>version: '3'

services:
 filebeat:
 image: docker.elastic.co/beats/filebeat:7.4.0
 volumes:
 - ./weblogs.log:/data/weblogs.log
 - ./filebeat.yml:/usr/share/filebeat/filebeat.yml
 depends_on:
 - kafka
 zookeeper:
 image: confluentinc/cp-zookeeper:5.3.2
 ports:
 - 2181:2181
 environment:
 ZOOKEEPER_CLIENT_PORT: 2181
 ZOOKEEPER_TICK_TIME: 2000
 kafka:
 image: confluentinc/cp-kafka:5.3.2
 depends_on:
 - zookeeper
 ports:
 - 9092:9092
 environment:
 KAFKA_BROKER_ID: 1
 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
 KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
 nifi:
 image: xemuliam/nifi:1.9
 ports:
 - 8080:8080
 - 8081:8081
 - 8443:8443</code></pre>
</li>
<li>
docker 컨테이너를 실행합니다.
<pre><code>docker-compose up -d</code></pre>
</li>
<li>
브라우저에서 nifi에 접속합니다. 접속까지 다소 오래 걸릴 수 있습니다. (브라우저 &gt; localhost:8080/nifi 접속)
<figure data-origin-width="0" data-origin-height="0" width="860" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/JD348/btqACPK7P1o/CRa95KaqlSyQjgEG69c33k/img.png" data-origin-width="0" data-origin-height="0" width="860" data-ke-mobilestyle="widthContent"></figure></li>
<li>
ConsumerKafka 프로세서를 추가하고, PROPERTIES 탭의 설정 값을 아래와 같이 변경합니다. (Processor 드래그 &gt; Filter에 kafka 검색 &gt; ADD 클릭)
<figure data-origin-width="0" data-origin-height="0" width="860" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/er6CHj/btqAKvEBNk6/tAkB6QUbOz9Tg4sKDfOrv1/img.png" data-origin-width="0" data-origin-height="0" width="860" data-ke-mobilestyle="widthContent"></figure>
<ul>
<li>Kafka Brokers: kafka:29092</li>
<li>Topic Name(s): filebeats-weblogs</li>
<li>Group ID: filebeats-weblogs-consumer</li>
</ul>
</li>
<li>
EvaluateJsonPath 프로세서를 추가하고, PROPERTIES 탭에서 우측 더하기 버튼으로 아래 설정 값을 추가합니다.
<figure width="860" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/cnnKfg/btqAK3HOJSk/TKDXGQin5BNztG0SbxCbx1/img.png" width="860" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"></figure>
<ul>
<li>Property Name: message</li>
<li>Value: $.message</li>
</ul>
</li>
<li>
PublishKafka 프로세서를 추가하고, PROPERTIES 탭의 설정 값을 아래와 같이 변경합니다.
<figure width="860" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/dPg0DQ/btqAJwjSvb2/8KV4sPy0CkTaaPZf5TEJ9k/img.png" width="860" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"></figure>
<ul>
<li>Kafka Brokers: kafka:29092</li>
<li>Topic Name: weblogs</li>
</ul>
</li>
<li>
각 프로세서를 드래그해서 연결합니다. EvaulatedJsonPath에서 PublishKafka로 연결할 때의 관계는 matched로 설정합니다. (ConsumerKafka &gt; EvaluateJsonPath &gt; PublishKafka 순서)
</li>
<li>
EvaluateJsonPath 프로세서와 PublishKafka 프로세서의 Automatically Terminate Relationships 옵션을 각각 아래와 같이 체크하고 저장해줍니다.
<ul>
<li>EvaluateJsonPath: failure, unmatched</li>
<li>PublishKafka: failure, success</li>
</ul>
</li>
<li>
각 프로세서를 시작해줍니다.
<figure width="320" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/tI571/btqALu6g6on/uk0fwf2gnWj1uJ4cXgOnB0/img.png" width="320" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"></figure></li>
</ol>
<h4 data-ke-size="size20">데이터 스트림 확인</h4>
<ol>
<li>
이제 데이터 소스를 실제와 같이 구현하기 위해 아래 명령어를 실행합니다. 해당 명령어는 fake log 생성기인 <a href="https://github.com/mingrammer/flog">flog</a>를 이용해 1초에 한 번씩 10개 row를 <code>weblogs.log</code> 파일에 더해주는 명령입니다.
<pre><code>while true; do docker run -it --rm mingrammer/flog:0.3.2 -n 10 &gt;&gt; weblogs.log; sleep 1; done</code></pre>
</li>
<li>
실제로 생성된 메세지를 확인하기 위해 터미널 탭을 2개 실행한 후, kafka 인스턴스에 접속합니다.
<pre><code>docker-compose exec kafka bash</code></pre>
</li>
<li>
각 탭에서 두 개의 consumer를 통해 실제로 메세지가 잘 흘러가고 있는지 확인합니다.
<pre><code>kafka-console-consumer --bootstrap-server localhost:9092 --topic filebeats-weblogs</code></pre>
<pre><code>kafka-console-consumer --bootstrap-server localhost:9092 --topic weblogs</code></pre>
</li>
<li>
nifi의 대시보드에서 kafka 프로세서 간 데이터 흐름을 확인합니다.
<figure width="320" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"><img src="https://blog.kakaocdn.net/dn/b2Uzji/btqAI33zHyM/2Jnl0RHzlKUIPadxP1HaHk/img.png" width="320" data-origin-width="0" data-origin-height="0" data-ke-mobilestyle="widthContent"></figure></li>
</ol>
 
 
											
					
 
 
 </div>

 
 
 
 
 
 


 

 
 
 
 </div></div>

생각나는대로 적어본다.

이번 포스트에서는 Filebeat, Kafka, Nifi를 docker-compose로 구성하고, 데이터 스트림을 생성하는 과정을 진행해보겠습니다. 튜토리얼 방식으로, 개인 로컬 맥에서 간단하게 따라서 진행해주시면 됩니다.
데이터는 Filebeat -> Kafka -> Nifi -> Kafka 순서로 흘러가는 구조입니다. 데이터 소스로 Apache 웹 로그를 사용해 실제 환경과 같이 일정 속도로 흘러가도록 구성하고, 실시간으로 데이터 스트림이 동작하는 모습을 확인해보겠습니다.
Tested on:
MacOS v10.14.6
docker v19.03
docker-compose v1.24
구성하기
적당한 workspace로 이동한 뒤, Filebeat에 제공하기 위한 데이터 소스로 weblogs.log 파일을 생성합니다. 일단 빈 파일로 생성하고, 추후에 데이터를 추가할 예정입니다.
touch weblogs.log
filebeat.yml 파일을 생성하고 아래 내용을 추가합니다.
filebeat:
  inputs:
  - type: log
    paths:
      - /data/weblogs.log
  config:
    modules:
      path: ${path.config}/modules.d/*.yml
      reload.enabled: false
output:
  kafka:
    hosts: ['kafka:29092']
    topic: 'filebeats-weblogs'
    partition:
      round_robin:
        reachable_only: false
    required_acks: 1
    compression: gzip
    max_message_bytes: 1000000
docker-compose.yml 파일을 생성하고 아래 내용을 추가합니다.
version: '3'
services:
  filebeat:
    image: docker.elastic.co/beats/filebeat:7.4.0
    volumes:
      - ./weblogs.log:/data/weblogs.log
      - ./filebeat.yml:/usr/share/filebeat/filebeat.yml
    depends_on:
      - kafka
  zookeeper:
    image: confluentinc/cp-zookeeper:5.3.2
    ports:
      - 2181:2181
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  kafka:
    image: confluentinc/cp-kafka:5.3.2
    depends_on:
    - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  nifi:
    image: xemuliam/nifi:1.9
    ports:
      - 8080:8080
      - 8081:8081
      - 8443:8443
docker 컨테이너를 실행합니다.
docker-compose up -d
브라우저에서 nifi에 접속합니다. 접속까지 다소 오래 걸릴 수 있습니다. (브라우저 > localhost:8080/nifi 접속)
ConsumerKafka 프로세서를 추가하고, PROPERTIES 탭의 설정 값을 아래와 같이 변경합니다. (Processor 드래그 > Filter에 kafka 검색 > ADD 클릭)
Kafka Brokers: kafka:29092
Topic Name(s): filebeats-weblogs
Group ID: filebeats-weblogs-consumer
EvaluateJsonPath 프로세서를 추가하고, PROPERTIES 탭에서 우측 더하기 버튼으로 아래 설정 값을 추가합니다.
Property Name: message
Value: $.message
PublishKafka 프로세서를 추가하고, PROPERTIES 탭의 설정 값을 아래와 같이 변경합니다.
Kafka Brokers: kafka:29092
Topic Name: weblogs
각 프로세서를 드래그해서 연결합니다. EvaulatedJsonPath에서 PublishKafka로 연결할 때의 관계는 matched로 설정합니다. (ConsumerKafka > EvaluateJsonPath > PublishKafka 순서)
EvaluateJsonPath 프로세서와 PublishKafka 프로세서의 Automatically Terminate Relationships 옵션을 각각 아래와 같이 체크하고 저장해줍니다.
EvaluateJsonPath: failure, unmatched
PublishKafka: failure, success
각 프로세서를 시작해줍니다.
데이터 스트림 확인
이제 데이터 소스를 실제와 같이 구현하기 위해 아래 명령어를 실행합니다. 해당 명령어는 fake log 생성기인 flog를 이용해 1초에 한 번씩 10개 row를 weblogs.log 파일에 더해주는 명령입니다.
while true; do docker run -it --rm mingrammer/flog:0.3.2 -n 10 >> weblogs.log; sleep 1; done
실제로 생성된 메세지를 확인하기 위해 터미널 탭을 2개 실행한 후, kafka 인스턴스에 접속합니다.
docker-compose exec kafka bash
각 탭에서 두 개의 consumer를 통해 실제로 메세지가 잘 흘러가고 있는지 확인합니다.
kafka-console-consumer --bootstrap-server localhost:9092 --topic filebeats-weblogs
kafka-console-consumer --bootstrap-server localhost:9092 --topic weblogs
nifi의 대시보드에서 kafka 프로세서 간 데이터 흐름을 확인합니다.

Filebeat-Kafka-Nifi로 데이터 스트림 5분 안에 구성해보기

댓글