Flume消费Kafka数据落盘至HDFS的实践

vlambda
2021-04-02

Flume消费Kafka数据落盘至HDFS的实践

# 需求背景

我司目前的日志采集是由 Structured Streaming 程序消费数据Kafka中的topic数据，经过ETL后写入HIVE表，但是该类日志程序占用约7个G的内存，这对于我们集群的压力来讲是比较大的(10台32g)。

# 组件调研

当前市面上的日志采集框架还是不少的

- Apache Flume

- Fluentd

- Logstash

- Chukwa

- Scribe

- Splunk Forwarder

经过分析后，我们决定采用flume作为消费者来消费Kafka中的数据。(flume对Kafka的支持还是不错的)

flume 经典的 source chnannel sink 组件

组件介绍

- Agent是一个JVM进程，它以事件的形式将数据从源头送至目的。

Agent主要有3个部分组成，Source、Channel、Sink。

- Source是负责接收数据到Flume Agent的组件。Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。

- Sink不断地轮询Channel中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个Flume Agent。

Sink组件目的地包括hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定义。

- Channel是位于Source和Sink之间的缓冲区。因此，Channel允许Source和Sink运作在不同的速率上。Channel是线程安全的，可以同时处理几个Source的写入操作和几个Sink的读取操作。

Flume自带两种Channel：Memory Channel和File Channel。

Memory Channel是内存中的队列。Memory Channel在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么Memory Channel就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。

File Channel将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。

- 传输单元，Flume数据传输的基本单元，以Event的形式将数据从源头送至目的地。Event由Header和Body两部分组成，Header用来存放该event的一些属性，为K-V结构，Body用来存放该条数据，形式为字节数组。

Flume消费Kafka数据落盘至HDFS的实践

# flume测试

## 安装

略

需要注意的是 flume在往hdfs导数据时，需要将hadoop相关jar包复制至flume安装目录中的lib包下。并确认Hadoop环境和Java环境配置正常 (检查/etc/profile.d/my_env.sh)

#JAVA_HOMEexport JAVA_HOME=/opt/module/jdk1.8.0_212export PATH=$PATH:$JAVA_HOME/bin

##HADOOP_HOMEexport HADOOP_HOME=/opt/module/hadoop-2.6.2export PATH=$PATH:$HADOOP_HOME/binexport PATH=$PATH:$HADOOP_HOME/sbin

注意：

如果你的Hadoop为3.1.x 版本

删除

flume/lib/guava-11.0.2.jar

以兼容Hadoop

## 案例实操

Flume消费Kafka数据落盘至HDFS的实践

上图为flume 官方案例式样

启动flume 实例需要新建一个example(自定义名字).conf文件

Flume消费Kafka数据落盘至HDFS的实践

官方的快速上手示例，有兴趣的同学可以自己动手尝试

Flume消费Kafka数据落盘至HDFS的实践

# flume实战

## source组件选型之Kafka Source

Flume消费Kafka数据落盘至HDFS的实践

## channel组件选型之File Channel

Flume消费Kafka数据落盘至HDFS的实践

## sink组件选型之HDFS Sink

Flume消费Kafka数据落盘至HDFS的实践

在了解各组件的基本配置说明之后

在flume的目录内新建job文件夹

mkdir job

cd job/ 

vi Kafka-flume-hdfs.conf

配置文件内容：

## 组件a1.sources=r1a1.channels=c1a1.sinks=k1

## source1# source 类型a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSourcea1.sources.r1.batchSize = 5000a1.sources.r1.batchDurationMillis = 2000# Kafka 地址a1.sources.r1.kafka.bootstrap.servers = xxx01:9092,xxx02:9092# Kafka 主题a1.sources.r1.kafka.topics=topic_test,topic_test2,topic_test3

a1.sources.r1.kafka.consumer.group.id = flume_kafka2hdfs

## channel1# channel 类型a1.channels.c1.type = filea1.channels.c1.maxFileSize = 2146435071a1.channels.c1.capacity = 1000000a1.channels.c1.keep-alive = 6

a1.channels.c1.checkpointDir = /opt/apache-flume-1.9.0-bin/datas/flumecheckpointDir1a1.channels.c1.dataDirs = /opt/apache-flume-1.9.0-bin/datas/flumedataDir1



## sink1# sink 类型a1.sinks.k1.type = hdfs# 数据文件在hdfs的路径a1.sinks.k1.hdfs.path = /tmp/flume/%{topic}/%Y-%m-%da1.sinks.k1.hdfs.filePrefix = log-a1.sinks.k1.hdfs.round = false

# 数据3600s滚动一次a1.sinks.k1.hdfs.rollInterval = 3600# 数据达到128M 滚动一次a1.sinks.k1.hdfs.rollSize = 134217728a1.sinks.k1.hdfs.rollCount = 0

## 控制输出文件是原生文件。(根据需要可以配置压缩)a1.sinks.k1.hdfs.fileType = DataStreama1.sinks.k1.hdfs.writeFormat=Text

## 拼装a1.sources.r1.channels = c1a1.sinks.k1.channel= c1

启动 flume实例

bin/flume-ng agent --conf conf --conf-file job/Kafka-flume-hdfs.conf --name a1 -Dflume.root.logger=INFO.console

相应的在Kafka 客户端启动三个生产者

 bin/kafka-console-producer.sh \--broker-test xxx01:9092 --topic test

bin/kafka-console-producer.sh \--broker-test xxx01:9092 --topic test2

bin/kafka-console-producer.sh \--broker-test xxx01:9092 --topic test3