kafka数据 落盘_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测试

安装

略

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

```

#JAVA_HOME

export JAVA_HOME=/opt/module/jdk1.8.0_212

export PATH=$PATH:$JAVA_HOME/bin

##HADOOP_HOME

export HADOOP_HOME=/opt/module/hadoop-2.6.2

export PATH=$PATH:$HADOOP_HOME/bin

export PATH=$PATH:$HADOOP_HOME/sbin

```

注意 ：

如果你的Hadoop为3.1.x 版本

删除

flume/lib/guava-11.0.2.jar

以兼容Hadoop

案例实操

上图为flume 官方案例式样

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

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

flume实战

source组件选型之Kafka Source

channel组件选型之File Channel

sink组件选型之HDFS Sink

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

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

```

mkdir job

cd job/

vi Kafka-flume-hdfs.conf

```

配置文件内容：

```

## 组件

a1.sources=r1

a1.channels=c1

a1.sinks=k1

## source1

# source 类型

a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource

a1.sources.r1.batchSize = 5000

a1.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 = file

a1.channels.c1.maxFileSize = 2146435071

a1.channels.c1.capacity = 1000000

a1.channels.c1.keep-alive = 6

a1.channels.c1.checkpointDir = /opt/apache-flume-1.9.0-bin/datas/flumecheckpointDir1

a1.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-%d

a1.sinks.k1.hdfs.filePrefix = log-

a1.sinks.k1.hdfs.round = false

# 数据3600s滚动一次

a1.sinks.k1.hdfs.rollInterval = 3600

# 数据达到128M 滚动一次

a1.sinks.k1.hdfs.rollSize = 134217728

a1.sinks.k1.hdfs.rollCount = 0

## 控制输出文件是原生文件。(根据需要可以配置压缩)

a1.sinks.k1.hdfs.fileType = DataStream

a1.sinks.k1.hdfs.writeFormat=Text

## 拼装

a1.sources.r1.channels = c1

a1.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

```

进行数据测试

正常接收数据。

Flume消费Kafka数据落盘至HDFS任务完成

经过监控得知内存消耗为500-700m ，相比较spark程序性能优化 7 倍。

优化任务完成。

最后推荐一位朋友的公众号，他会在业余时间输出一些Java相关的知识。