[Kafka] 7. Apache Flume

Apache Flume

Apache Flume은 오픈소스 프로젝트로 개발된 로그 데이터를 수집 기술이다. 여러 서버에서 생산된 대용량 로그 데이터를 효과적으로 수집하여, HDFS과 같은 원격 목적지에 데이터를 전송하는 기능을 제공한다. 구조가 단순하고 유연하여 다양한 유형의 스트리밍 데이터 플로우(Streaming Data Flow) 아키텍처를 구성할 수 있다.

• flume은 Source, Channel, Sink 세 가지 요소로 구성
  • 데이터가 추출되는 Source
  • Flume에 데이터를 저장하는 Sink
  • Source에서 Sink or 저장소로 데이터를 전달하는 Channel

Flume 설치

• https://flume.apache.org/download.html 사이트에 들어가서 다운받을 수 있다.

구현

• source-topic과 target-topic을 생성

1. flume의 confi 폴더 안에 flume.conf 파일을 생성

flume1을 flume 인스턴스로 선언하고 kafka-source-1, mem-channel-1, kafka-sink-1로 설정한다.

flume1.sources = kafka-source-1
flume1.channels = mem-channel-1
flume1.sinks = kafka-sink-1

소스의 설정을 선언하는 내용.

## type: 소스타입 설정
flume1.sources.kafka-source-1.type=org.apache.flume.source.kafka.KafkaSource
## zookeeperConnect : 주키퍼의 연결 문자열이며, host:port 형슥올 구분해서 지정한다.
flume1.sources.kafka-source-1.zookeeperConnect = localhost:2181
## topoic 읽어올 소스를 지정한다. flume은 기록할 때 소스당 오직 한 개의 kafka topic을 지원한다.
flume1.sources.kafka-source-1.topic = source-topic
## batchSize : kafka에서 메시지를 가져와서 채널에 기록할 최대 메시지 수이다.(default : 1000) 이 값은 한 번 가져올 때 채널이 처리할 수 있는 데이터양에 따라 결정된다.
flume1.sources.kafka-source-1.batchSize = 100
## channels : source를 연결할 channel 설정
flume1.sources.kafka-source-1.channels = mem-channel-1

## 그외
## batchDurationMillis : 시스템이 배치(batch)를 채널에 기록하기 전에 시스템이 대기할 최대 시간을 밀리초 단위로 지정한다. batchSize가 이시간이 되기전에 초과되면 배치는 해당채널로 전송한다. (default: 1000)

source와 저장소 사이의 channel을 정의.

## 메모리가 데이터를 보관하는데 사용되므로 메모리 채널로 설정했다.
## type : 메모리 채널 사용을 설정하기 위해 memory로 설정.
## 채널유형 : memory, JDBC file, kafka channel
flume1.channels.mem-channel-1.type = memory

##그외
## capacity : 메모리에 보관이 가능한 최대 메시지 수. 메모리 용량과 메시지 크기를 고려하여 설정 (default : 100)
## transactionCapacity : 소스에서 가져오거나 저장소에 하나의 트랜잭션으로 처리할 최대 메시지 수

저장소 설정을 선언하는 내용

## type : 저장소유형 지정
flume1.sinks.kafka-sink-1.type = org.apache.flume.sink.kafka.KafkaSink
## brokerList : 메시지를 기록할 kafka cluster broker list이다. host:port 형식으로 여러개 입력가능
flume1.sinks.kafka-sink-1.brokerList = localhost:9092
## 메시지를 기록할 kafka topic
flume1.sinks.kafka-sink-1.topic = target-topic
## 한번에 기록할 메시지 수 지정
flume1.sinks.kafka.sink-1.batchSize = 50
## 데이터를 수집하기 위해 사용할 채널의 이름
flume1.sinks.kafka-sink-1.channel = mem-channel-1

[flume.conf]

flume1.sources = kafka-source-1
flume1.channels = mem-channel-1
flume1.sinks = kafka-sink-1


flume1.sources.kafka-source-1.type=org.apache.flume.source.kafka.KafkaSource
flume1.sources.kafka-source-1.zookeeperConnect = localhost:2181
flume1.sources.kafka-source-1.topic = source-topic
flume1.sources.kafka-source-1.batchSize = 100
flume1.sources.kafka-source-1.channels = mem-channel-1

flume1.channels.mem-channel-1.type = memory

flume1.sinks.kafka-sink-1.type = org.apache.flume.sink.kafka.KafkaSink
flume1.sinks.kafka-sink-1.brokerList = localhost:9092
flume1.sinks.kafka-sink-1.topic = target-topic
flume1.sinks.kafka.sink-1.batchSize = 50
flume1.sinks.kafka-sink-1.channel = mem-channel-1

2. target-topic에 데이터를 푸시하는 flume 에이전드 실행

$ flume-ng agent --conf-file flume.conf --name flume1