Apache Flume 연동

Apahe Flume은 분산 환경에서 대량의 로그 데이터를 효과적으로 수집하여 데이터 저장소로 전송할 수 있는 서비스입니다.
자세한 사항은 Flume 공식 홈페이지를 참고해 주십시오.

hadoop-chadoop-use-ex8_0-1

성격
- Distributed: 토폴로지(Topology)를 어떻게 구성하느냐에 따라 달라질 수 있지만 여러 개의 Flume Agent끼리 파이프를 만들 수 있습니다. 보통 Agent끼리 연결할 때에는 Avro 타입의 Sink, Source(next hop)를 사용합니다.
- Reliable: Flume Agent 안에서 이벤트는 Source, Channel, Sink라는 컴포넌트를 따라서 이동합니다. 이벤트가 Sink로 전달되기 전까지는 Channel에서 이벤트가 사라지지 않고 이를 보장하기 위해 Flume은 Transactional approach를 도입했습니다.
- Available: Channel로 Disk-backed 시스템을 사용한다면 Agent에 오류가 발생해도 Source에서 Channel로 전달된 데이터를 복구할 수 있습니다.
용도
- rom many ~ to a centralized: Flume Agent는 여러 개의 노드에서 로그를 가져와 최종적으로는 중앙화된 저장소에 보관할 수 있습니다.
- collecting, aggregating, moving: 로그를 수집하여 합칠 수 있습니다. 그 과정 중에 Selector, Interceptor를 활용하여 이벤트의 형태를 바꿀 수 있습니다.
  수집한 이벤트는 다음 Agent로 전달하거나 최종 Sink에 저장할 수 있습니다.
컴포넌트
- Event: Flume Agent에 의해 옮겨지는 데이터의 기본 단위입니다. 선택적으로 이벤트에는 헤더값을 줄 수 있으며 보통 헤더는 이벤트의 내용을 확인하고 변경하기 위해 사용합니다.
- Flume Agent: JVM 프로세스로 Source, Channel, Sink 컴포넌트를 호스트 합니다. 이벤트는 Agent를 통해서 외부의 Source로부터 next-hop의 목적지로 흘러갈 수 있습니다.
  - Source: 클라이언트로 부터 전달받은 이벤트를 소비합니다. Source가 이벤트를 받으면 1개 이상의 채널에 넘기게 됩니다.
  - Channel: 이벤트의 임시 저장소입니다. Source와 Sink를 이어주는 역할을 하며 이벤트 흐름의 Durability를 보장하는 데 중요한 역할을 합니다.
  - Sink: 채널로부터 이벤트를 제거하고 flow의 next-hop에 전달합니다.

이 가이드에서는 Cloud Hadoop의 HDFS에 서버의 로그를 저장하는 Flume 토폴로지 구성 방법을 설명합니다.

Flume Agent 사용

Flume Agent를 사용하여 아래와 같이 각 서버의 vmstat 결과를 수집하여 Cloud Hadoop HDFS에 저장하는 Flume 토폴로지를 구성해 볼 수 있습니다.
hadoop-chadoop-use-ex8_1-1

Flume 설치

Flume Agent를 설치하는 방법은 다음과 같습니다.

로그를 수집할 서버 3대를 생성해 주십시오. (Server 생성 참고)
- 각 서버는 Cloud Hadoop이 포함된 ACG에 생성해야 합니다.
- log-gen-001 / centos-7.8–64 / 2vCPU, 8G Mem
- log-gen-002 / centos-7.8–64 / 2vCPU, 8G Mem
- log-gen-003 / centos-7.8–64 / 2vCPU, 8G Mem

~/downloads ~/apps 경로의 디렉터리를 생성한 후 해당 경로로 Flume 패키지를 다운로드해 압축을 풀고 설치를 완료해 주십시오.

mkdir ~/downloads ~/apps

cd ~/downloads
wget https://archive.apache.org/dist/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz
tar -xvf apache-flume-1.9.0-bin.tar.gz

mv apache-flume-1.9.0-bin ~/apps/
cd ~/apps
ln -s apache-flume-1.9.0-bin flume

사전 작업

Cloud Hadoop을 HDFS Sink로 쓰려면 아래와 같은 사전 작업이 필요합니다.

1. 웹 서버와 HDFS 통신을 위한 사전 작업

각 로그 서버와 HDFS 네임노드 호스트와 통신할 수 있도록 각 호스트의 Private IP와 호스트명을 /etc/hosts에 등록해 주십시오.

해당 정보는 Cloud Hadoop의 엣지 노드 (e.g. e-001-xxx )의 /etc/hosts에서 확인할 수 있습니다.

2. HDFS Sink를 사용하기 위한 사전 작업

HDFS Sink를 사용하기 위해 이 토폴로지에서는 HDFS Sink를 사용합니다. Flume Agent가 실행되는 노드에서 Hadoop common jar 라이브러리가 필요하게 됩니다. NameNode HA를 위해 네임서비스를 사용하려면 hdfs-site.xml , core-site.xml 같은 구성 파일도 필요합니다.

Hadoop 바이너리 다운로드
다음 명령어를 사용하여 /home 아래에서 필요한 Hadoop 바이너리와 .jar 라이브러리를 다운로드해 주십시오.

Cloud Hadoop 1.3 버전에서는 아래 코드를 사용해 주십시오.

# HADOOP 바이너리 다운로드
wget https://archive.apache.org/dist/hadoop/common/hadoop-2.6.5/hadoop-2.6.5.tar.gz -P ~/apps
tar xfz hadoop-2.6.5.tar.gz  

# HDFS Jar 다운로드
wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-auth/2.6.5/hadoop-auth-2.6.5.jar -P ~/apps/hadoop-2.6.5/share/hadoop/common
wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-hdfs/2.6.5/hadoop-hdfs-2.6.5.jar -P ~/apps/hadoop-2.6.5/share/hadoop/common
wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-hdfs-client/2.6.5/hadoop-hdfs-client-2.6.5.jar -P ~/apps/hadoop-2.6.5/share/hadoop/common

Cloud Hadoop 1.4 버전 이상에서는 아래 코드를 사용해 주십시오.

# HADOOP 바이너리 다운로드
wget https://archive.apache.org/dist/hadoop/common/hadoop-3.1.4/hadoop-3.1.4.tar.gz -P ~/apps
tar xfz hadoop-3.1.4.tar.gz  

# HDFS Jar 다운로드
wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-auth/3.1.4/hadoop-auth-3.1.4.jar -P ~/apps/hadoop-3.1.4/share/hadoop/common
wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-hdfs/3.1.4/hadoop-hdfs-3.1.4.jar -P ~/apps/hadoop-3.1.4/share/hadoop/common
wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-hdfs-client/3.1.4/hadoop-hdfs-client-3.1.4.jar -P ~/apps/hadoop-3.1.4/share/hadoop/common

Hadoop config 설정
$FLUME_CLASS_PATH/conf 아래 Hadoop config 파일을 다운로드해 주십시오.

$ cd ~/apps/flume/conf
$ curl -u $AMBARI_ID:$AMBARI_PASS -G 'http://$AMBARI_URI:8080/api/v1/clusters/$CLUSTER_NAME/components?format=client_config_tar' -o client_config.tgz 
$ tar xfz client_config.tgz
$ rm -f client_config.tgz

Hadoop 환경 변수 설정
다음 명령어를 실행하여 Hadoop 환경 변수를 설정해 주십시오.

export HADOOP_HOME=~/apps/hadoop-3.1.4
export HADOOP_HDFS_HOME=$HADOOP_HOME
export HADOOP_CONF_DIR=~/apps/flume/conf/HDFS_CLIENT/
export PATH=${JAVA_HOME}/bin:${HADOOP_HOME}/bin:${PATH}

Flume 구성 변경

Flume 구성을 변경하는 방법은 다음과 같습니다.

Flume Agent에서 다음 명령어를 실행하여 구성값을 생성해 주십시오.

cd ~/apps/flume/conf
cp flume-conf.properties.template flume.conf
cp flume-env.sh.template flume-env.sh
cp ~/apps/hadoop-3.1.4/share/hadoop/common/*.jar ~/apps/flume/lib/ 
cp ~/apps/hadoop-3.1.4/share/hadoop/common/lib/woodstox-core-5.0.3.jar ~/apps/flume/lib/
mv ~/apps/flume/lib/guava-11.0.2.jar ~/apps/flume/lib/guava-11.0.2.jar.bak

각 Flume Agent에서 아래처럼 hadoop-env.sh 의 JAVA_HOME과 HADOOP_HOME 옵션값을 수정해 주십시오.
- Java 설정은 설치 방법에 따라 옵션값이 다를 수 있습니다. 아래는 yum으로 Java 패키지를 설치한 후 설정한 경로입니다.

# Java 설치
yum install -y java-1.8.0-openjdk

# hadoop-evn.sh 편집
vi $HADOOP_HOME/etc/hadoop/hadoop-env.sh

# The java implementation to use.  Required.
export JAVA_HOME=/usr/lib/jvm/jre-openjdk
# Hadoop home directory
export HADOOP_HOME=/root/apps/hadoop-3.1.4

flume.conf

Agent의 이름과 각 컴포넌트를 정의합니다. (Agent 이름: fooAgent)
HDFS의 Sink 경로에는 네임서비스가 들어간 경로를 사용하면 됩니다. Cloud Hadoop에서는 클러스터명이 네임서비스가 됩니다.
hdfs-site.xml에 노드 정보가 포함되어 있으므로 어느 네임노드가 Active 상태인지 명시하지 않아도 됩니다.

fooAgent.sources = Exec
fooAgent.channels = MemChannel
fooAgent.sinks = HDFS

fooAgent.sources.Exec.type = exec
fooAgent.sources.Exec.command = /usr/bin/vmstat 1
fooAgent.sources.Exec.channels = MemChannel 

fooAgent.channels.MemChannel.type = memory
fooAgent.channels.MemChannel.capacity = 10000
fooAgent.channels.MemChannel.transactionCapacity = 1000

fooAgent.sinks.HDFS.channel = MemChannel
fooAgent.sinks.HDFS.type = hdfs
fooAgent.sinks.HDFS.hdfs.path = hdfs://$CLUSTER_NAME/user/hduser/flume/events/
fooAgent.sinks.HDFS.hdfs.fileType = DataStream
fooAgent.sinks.HDFS.hdfs.writeFormat = Text
fooAgent.sinks.HDFS.hdfs.batchSize = 1000
fooAgent.sinks.HDFS.hdfs.rollSize = 0
fooAgent.sinks.HDFS.hdfs.rollCount = 10000

flume-env.sh
사전 작업에서 설치한 hadoop client의 경로를 FLUME_CLASSPATH에 추가합니다.

export JAVA_HOME="/usr/lib/jvm/jre-openjdk"
export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"
export HADOOP_CONF_DIR="/root/apps/flume/conf/HDFS_CLIENT"
FLUME_CLASSPATH="/root/apps/flume/lib"

프로세스 시작

디렉터리 생성 및 소유자 권한 설정

$ sudo su - hdfs
$ hdfs dfs -mkdir /user/hduser/flume/events/
$ hdfs dfs -chown -R sshuser: /user/hduser/flume/events/
$ exit

다음 명령어를 사용하여 각 Flume Agent를 시작해 주십시오.

cd ~/apps/flume/

./bin/flume-ng agent --conf ./conf/ -f conf/flume.conf -Dflume.root.logger=DEBUG,console -n  fooAgent

....

20/09/09 12:32:22 INFO hdfs.HDFSDataStream: Serializer = TEXT, UseRawLocalFileSystem = false
20/09/09 12:32:22 INFO hdfs.BucketWriter: Creating hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622342911.tmp
20/09/09 12:32:52 INFO hdfs.HDFSEventSink: Writer callback called.
20/09/09 12:32:52 INFO hdfs.BucketWriter: Closing hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622342911.tmp
20/09/09 12:32:52 INFO hdfs.BucketWriter: Renaming hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622342911.tmp to hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622342911
20/09/09 12:32:55 INFO hdfs.HDFSDataStream: Serializer = TEXT, UseRawLocalFileSystem = false
20/09/09 12:32:55 INFO hdfs.BucketWriter: Creating hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622375913.tmp
20/09/09 12:33:25 INFO hdfs.HDFSEventSink: Writer callback called.
20/09/09 12:33:25 INFO hdfs.BucketWriter: Closing hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622375913.tmp
20/09/09 12:33:25 INFO hdfs.BucketWriter: Renaming hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622375913.tmp to hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622375913
20/09/09 12:33:28 INFO hdfs.HDFSDataStream: Serializer = TEXT, UseRawLocalFileSystem = false
20/09/09 12:33:28 INFO hdfs.BucketWriter: Creating hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622408915.tmp
20/09/09 12:33:58 INFO hdfs.HDFSEventSink: Writer callback called.
20/09/09 12:33:58 INFO hdfs.BucketWriter: Closing hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622408915.tmp
20/09/09 12:33:58 INFO hdfs.BucketWriter: Renaming hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622408915.tmp to hdfs://xxxxx/user/hduser/flume/events/FlumeData.1599622408915
20/09/09 12:34:01 INFO hdfs.HDFSDataStream: Serializer = TEXT, UseRawLocalFileSystem = false
20/09/09 12:34:01 INFO hdfs.BucketWriter: Creating

아래 명령어를 사용하여 HDFS에서 확인할 수 있습니다.

$ hadoop fs -ls /user/hduser/flume/events/

Found 17 items

-rw-r--r--   2 root hdfs       3089 2020-09-09 12:25 /user/hduser/flume/events/FlumeData.1599621914876
-rw-r--r--   2 root hdfs       3093 2020-09-09 12:26 /user/hduser/flume/events/FlumeData.1599621946882
-rw-r--r--   2 root hdfs       2931 2020-09-09 12:26 /user/hduser/flume/events/FlumeData.1599621979885
-rw-r--r--   2 root hdfs       3091 2020-09-09 12:27 /user/hduser/flume/events/FlumeData.1599622012888
-rw-r--r--   2 root hdfs       2931 2020-09-09 12:27 /user/hduser/flume/events/FlumeData.1599622045890
-rw-r--r--   2 root hdfs       3091 2020-09-09 12:28 /user/hduser/flume/events/FlumeData.1599622078893
-rw-r--r--   2 root hdfs       2930 2020-09-09 12:29 /user/hduser/flume/events/FlumeData.1599622111895
-rw-r--r--   2 root hdfs       3093 2020-09-09 12:29 /user/hduser/flume/events/FlumeData.1599622144897
-rw-r--r--   2 root hdfs       3092 2020-09-09 12:30 /user/hduser/flume/events/FlumeData.1599622177899
-rw-r--r--   2 root hdfs       2931 2020-09-09 12:30 /user/hduser/flume/events/FlumeData.1599622210902
-rw-r--r--   2 root hdfs       3093 2020-09-09 12:31 /user/hduser/flume/events/FlumeData.1599622243904
-rw-r--r--   2 root hdfs       2932 2020-09-09 12:31 /user/hduser/flume/events/FlumeData.1599622276906

참고

실제 운영 환경에서는 2개 이상의 Flume Agent를 파이프(pipe)하면서 인터셉터(interceptor)를 활용하여 이벤트를 변환합니다.
Source, Channel, Sink에는 다양한 타입이 존재하며 Kafka를 Channel, Sink로 많이 사용합니다.