Kafka总结（七）：数据采集应用

水深无声 2022-05-18 04:41 654阅读 0赞

[Kafka总结（一）：Kafka概述][Kafka_Kafka]

[Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1]

[Kafka总结（三）：Kafka核心流程分析][Kafka_Kafka 2]

[Kafka总结（四）：Kafka命令操作][Kafka_Kafka 3]

[Kafka总结（五）：API编程详解][Kafka_API]

[Kafka总结（六）：Kafka Stream详解][Kafka_Kafka Stream]

[Kafka总结（七）：数据采集应用][Kafka]

[Kafka总结（八）：KafKa与ELK整合应用][Kafka_KafKa_ELK]

[Kafka总结（九）：KafKa 与Spark整合应用][Kafka_KafKa _Spark]

## 1.Log4j集成KafKa应用 ##

### 1.应用描述 ###

一般都是将日子保存在本地服务器中，如果将应用日志同步写在远程服务器，则会导致应用程序依赖于远程服务器，如果远程服务器宕机或者网络请求超时都会影响到应用程序，同时也会增加应用程序的网络开销。

因此，如果要将日志写到远程服务器，一定需要结合应用场景综合的考虑；

### 2.具体实现 ###

应用程序可以通过Log4j将日志同步到Kafka步骤：

1.  依赖：kafka-log4j-appender
2.  编写log4j.properties文件，为了保证不同级别的日志分别输出到不同的文件，并且每个级别的日志不再包含比该级别高的日志，可以自定义一个Appender，继承DailyRollingFileAppender类，重写isAsServerAsThresold()方法，让每个级别类型的日志只提取与之优先级（Priority）相等的日志；
3.  在Kafka上创建log4j.properties中配置的相应主题；

## 2. KafKa与Flume整合应用 ##

kafka-log4j-appender将应用程序的日志直接写入到kafka中，这种方式虽然简单，但是由于应用程序依赖于Kafka运行环境，因此存在一定的不足；

通常的做法是：应用程序将日志写入到本地，然后通过日志采集工具将本地日志同步到远程服务器，而Flume就是最常用的数据采集工具之一。现在引入了Flume，通过Flume将应用程序产生的日志同步到Kafka；

### 1. Flume简介 ###

![70][]

### 2. Flume与KafKa比较 ###

### 3. Flume的安装配置 ###

### 4. Flume采集日志写入KafKa ###

步骤：

1.  需要在配置文件中指定源、接收器和通道的名称；
2.  配置源
3.  配置通道信息
4.  配置接收器，接收器从通道获取信息写入到Kafka

## 3.KafKa与Flume和HDFS整合应用 ##

Flume是一个连接各种组件和系统的桥梁，下载安装的lib目录下有Flume与Hbase、HDFS等集成的jar文件，可以很方便的与HBase和HDFS连接。在实际业务中，我们一般通过Flume从应用程序采集实时数据写入到Kafka，而将历史数据写入到HBASE和HDFS。

例如：我们可以通过Kafka Steams实时对用户行为分析，将分析结果再写入到Kafka，然后有Flume写入Hbase，供可视化应用系统展示。通过将这些系统整合在一起，可以很方便的实现一个**扩展性良好、高吞吐量的数据采集分析系统；**

### 1.Hadoop安装配置 ###

步骤：

1.  解压Hadoop安装包
2.  环境配置：

1.  指定JDK安装路径，在hadoop-env.sh文件中添加JDK安装路径
2.  修改core-site.xml文件，在该文件中配置通信地址以及存储路径；
3.  修改hdfs-site.xml，在该文件中配置Hadoop文件块的数据备份数，由于是伪分布式模式，因此可以将其值设置为1，同时配置NameNode和DataNode节点文件块数据存储路径；
4.  修改mapred-site.xml。将配置模板文件mapred-site.xml.template复制一份，命令为mapred-site.xml，在该文件中配置JobTracker的主机名或者IP和端口。对于伪分布式模式该节点既是JobTracker也是TaskTracker。

1.  启动运行。在运行Hadoop之前先进性格式化Hadoop工作空间。

### 2.Flume采集KafKa消息写入HDFS ###

整合步骤：

1.  KafkaSource配置；

Flume提供了从Kafka采集数据的flume-kafka-source-1.0.7.jar文件，需要为该jar提供的KafkaSource进行相关的配置；

1.  KafkaChannel配置；

Flume采集数据的时候都采用内存通道，现在我们使用KafkaChannel，在安装lib目录下，有一个flume-kafka-channel-1.7.0.jar，该文件定义了一个Kafka通道，将采集的数据写入到Kafka主题中；

1.  写HDFS接收器配置；

在安装lib目录下有一个flume-hdfs-sink-1.7.0.jar文件，定义了一个将数据写入到HDFS的接入器HDFSEventSink，可以通过该接收器提供的配置写入HDFS**数据格式**、**文件切割方式**、**文件压缩方式**等；

[Kafka_Kafka]: https://mp.csdn.net/postedit/81283142
[Kafka_Kafka 1]: https://mp.csdn.net/postedit/81283229
[Kafka_Kafka 2]: https://mp.csdn.net/postedit/81283286
[Kafka_Kafka 3]: https://mp.csdn.net/postedit/81283397
[Kafka_API]: https://mp.csdn.net/postedit/81283460
[Kafka_Kafka Stream]: https://mp.csdn.net/postedit/81283491
[Kafka]: https://mp.csdn.net/postedit/81283546
[Kafka_KafKa_ELK]: https://mp.csdn.net/postedit/81283568
[Kafka_KafKa _Spark]: https://mp.csdn.net/postedit/81283606
[70]: /images/20220518/dcdd94229d0f436eb2a954aaf34ed085.png