flume multi sink 설정하기

단일 싱크 구조는 아래와 같다.

하나의 source와 channel, sink를 가지는 구조이다.

위 그림에서는 Flume을 이용하여 MongoDB에 저장하는 구조이다.

여기서 하나의 Event에 대해 2군데 데이터를 전송하려면 어떻게 해야할까?

예를 들어 하나의 Event를 보냈을 때, MongoDB에도 저장하고 HDFS 에도 저장하려고 한다.

2개의 sink를 연결해 볼 수 있다.

이 경우에는 Event 정보가 MongoDB나 HDFS 둘 중에 하나만 저장이 된다.

어떤 Sink가 선택이 되는지는 알 수 없으나, 설정상 나중에 연결된 sink 부터 전달되는 듯 하다.

테스트를 해보면 위 경우 HDFS Sink를 먼저 시도한다. 이 때 HDFS sink에 실패한다면 Mongo Sink를 시도한다

이는 조금만 생각해보면 당연한 것 같다.

channel 의 역할은 Event를 잠시 담아두는 Queue 역할이다. 하나의 Event만 들어있기 때문에 Sink도 한 쪽에만 되어야 한다.

즉, Sink 대상이 되는 Event가 한 개이니, 하나만 되는게 맞다.

그렇다면 양쪽에 Event를 Sink 하려면 어떻게 구조를 가져가면 될까?

아래와 같은 구조를 생각해 볼 수 있다.

channel에서 sink로 전달 될 때, 하나의 sink로만 Event가 전달되는데,

source에서 channel에 전달 될 때는 Event가 양쪽으로 전달될까? 하는 의문을 가질 수 있다.

https://flume.apache.org/FlumeUserGuide.html#multiplexing-the-flow

apache flume 가이드에 보면 multiplexing 설정이 있다.

하나의 source에 여러 개의 channel이 연결되는 구조이다.

이 때 연동되는 방식이 2가지가 있다.

1. multiplexing 방식

이 방식은 여러 개의 channel 중에 Event가 전달될 channel을 선택하여 전달하는 방식이다.

즉, 하나의 Event가 하나의 channel에만 전달할 수도 있고, 여러 channel에 전달할 수도 있다.

설정은 아래와 같이 할 수 있다.

agent.sources = source1 agent.sinks = mongoSink hdfsSink agent.channels = mongoChannel hdfsChannel agent.sources.source1.channels = mongoChannel hdfsChannel    # source1에 channel 2개를 연결한다. # 각 channel을 sink에 연결한다. agent.sinks.mongoSink.channel = mongoChannel agent.sinks.hdfsSink  .channel =hdfsChannel agent.sources.source1.selector.type = multiplexing agent.sources.source1.selector.header = transferType   # transferType은 예시이다. agent.sources.source1.selector.mapping.all = mongoChannel hdfsChannel agent.sources.source1.selector.mapping.mongo = mongoChannel agent.sources.source1.selector.mapping.hdfs = hdfsChannel

이렇게 설정할 경우 Event 전달시 header에 있는 transferType이라는 값을 체크하여

all 인 경우 mongo와 hdfs 양쪽에 보내고,

mongo인 경우 mongo만, hdfs인 경우 hdfs 만 보내게 된다.

2. replicating 방식

이 방식은 하나의 Event를 연결된 여러 개의 channel 모두에게 전달하는 방식이다.

즉, 위 예제대로면 mongoChannel과 hdfsChannel 양쪽에 Event가 전달되는 것이다.(Event를 양쪽 channel에 put하는 듯 하다.)

설정은 아래와 같이 할 수 있다.

agent.sources = source1 agent.sinks = mongoSink hdfsSink agent.channels = mongoChannel hdfsChannel agent.sources.source1.channels = mongoChannel hdfsChannel    # source1에 channel 2개를 연결한다. # 각 channel을 sink에 연결한다. agent.sinks.mongoSink.channel = mongoChannel agent.sinks.hdfsSink  .channel =hdfsChannel agent.sources.source1.selector.type = replicating

이렇게 설정할 경우 Event가 mongo와 hdfs 양쪽에 sink 되는 것을 확인할 수 있다.