Flume+Kafka+Storm实战:二、Flume与Kafka整合

简介: Flume+Kafka+Storm实战:二、Flume与Kafka整合
+关注继续查看

0x00 文章内容


  1. Flume准备
  2. Kafka准备
  3. 校验结果


PS:请自行准备好Flume、Kafka的环境。由于本教程是属于整合教程,所以,我们可以直接在原来的基础上进行升级即可。过程是将教程:Flume入门案例之NetCat-Souces里的Sink修改为Kafka,而这里的Kafka用的其实是教程:Flume+Kafka+Storm实战:一、Kakfa与Storm整合里面的topic。


0x01 Flume准备


1. 编写Flume配置文件

a. 拷贝一份example

cd ~/bigdata/apache-flume-1.8.0-bin

cp conf/example.conf conf/kafka.conf

b. 然后修改Sink(点击跳转官网参考

vi conf/kafka.conf


image.png


image.png


c. 完整配置文件

a1.sources = r1
a1.sinks = k1
a1.channels = c1

a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = word-count-input
a1.sinks.k1.kafka.bootstrap.servers = master:9092
a1.sinks.k1.kafka.flumeBatchSize = 5

a1.channels.c1.type = memory

a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1


0x02 Kafka准备


1. 创建topic(如已操作过可跳过)

a. 启动Kafka与ZK(三台都执行)

启动ZK:

zkServer.sh start

启动Kafka:

nohup ~/bigdata/kafka_2.11-1.0.0/bin/kafka-server-start.sh ~/bigdata/kafka_2.11-1.0.0/config/server.properties >~/bigdata/kafka_2.11-1.0.0/logs/server.log 2>&1 &


目前的进程情况(call_all.sh脚本请参考:大数据常用管理集群脚本集合):


image.png


b. 创建Topic:word-count-input(如果已创建则忽略)

查看Topic(在上一教程已经创建):

~/bigdata/kafka_2.11-1.0.0/bin/kafka-topics.sh --list --zookeeper master:2181


image.png


如果没创建则创建:

~/bigdata/kafka_2.11-1.0.0/bin/kafka-topics.sh --create --zookeeper master:2181 --replication-factor 1 --partitions 1 --topic word-count-input


0x03 校验结果


1. 启动Flume


a. 在master启动

bin/flume-ng agent --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/kafka.conf --name a1


image.png


2. 启动Kafka消费者

a. 首先应确保先将Kafka与ZK启动起来(已启动则忽略,三台都执行)

启动ZK:

zkServer.sh start

启动Kafka:

nohup ~/bigdata/kafka_2.11-1.0.0/bin/kafka-server-start.sh ~/bigdata/kafka_2.11-1.0.0/config/server.properties >~/bigdata/kafka_2.11-1.0.0/logs/server.log 2>&1 &

b. 在master启动Kafka消费者

kafka-console-consumer.sh --bootstrap-server master:9092 --topic word-count-input


3. 测试结果

a. 测试过程与教程:Flume入门案例之NetCat-Souces 一样,可以看到我们的内容可以在Kafka的消费者端接收到。


image.png


image.png


0xFF 总结


其实内容也比较简单,重点是要结合官网来进行配置,其余的都是学习过的内容。

实战的上一篇教程:Flume+Kafka+Storm实战:一、Kakfa与Storm整合,学习没有先后顺序,但是应该具备Flume、Kafka等基础。

不同的版本,配置与结果都有可能不同,关于flumeBatchSize的配置项问题,请参考文章:Flume1.7及以上版本的Kafka Sink batchsize(flumeBatchSize) 配置问题,flumeBatchSize为分批次进行发送配置。

Flume1.6.0版本参考:


先测试能否输出到控制台再对接Spark Streaming
技术选型一:avro-memory-logger
技术选型二:avro-memory-kafka

参考:
技术选型一:
#streaming_logger.conf
agent1.sources=avro-source
agent1.channels=logger-channel
agent1.sinks=log-sink
#define source
agent1.sources.avro-source.type=avro
agent1.sources.avro-source.bind=localhost
agent1.sources.avro-source.port=41414
#define channel
agent1.channels.logger-channel.type=memory
#define sink
agent1.sinks.log-sink.type=logger
agent1.sources.avro-source.channels=logger-channel
agent1.sinks.log-sink.channel=logger-channel

flume-ng agent \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/streaming_logger.conf \
--name agent1 \
-Dflume.root.logger=INFO,console

参考:
技术选型二:
#streaming_kafka.conf
agent1.sources=avro-source
agent1.channels=logger-channel
agent1.sinks=kafka-sink
#define source
agent1.sources.avro-source.type=avro
agent1.sources.avro-source.bind=0.0.0.0
agent1.sources.avro-source.port=41414
#define channel
agent1.channels.logger-channel.type=memory
#define sink
agent1.sinks.kafka-sink.type=org.apache.flume.sink.kafka.KafkaSink
agent1.sinks.kafka-sink.topic = streaming_topic
agent1.sinks.kafka-sink.brokerList = localhost:9092
agent1.sinks.kafka-sink.requiredAcks = 1
agent1.sinks.kafka-sink.batchSize = 20
agent1.sources.avro-source.channels=logger-channel
agent1.sinks.kafka-sink.channel=logger-channel

执行脚本:
flume-ng agent \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/streaming_kafka.conf \
--name agent1 \
相关文章
|
2天前
|
消息中间件 关系型数据库 MySQL
使用Flume实现MySQL与Kafka实时同步
使用Flume实现MySQL与Kafka实时同步
9 0
|
2月前
|
消息中间件 数据采集 分布式计算
flume kafka和sparkstreaming整合
flume kafka和sparkstreaming整合
|
2月前
|
消息中间件 存储 分布式计算
Flume实现Kafka数据持久化存储到HDFS
Flume实现Kafka数据持久化存储到HDFS
99 0
|
3月前
|
消息中间件 数据采集 存储
大数据数据采集的数据采集(收集/聚合)的Flume之数据采集流程的Channel的Kafka Channel
在大数据处理和管理中,数据采集是非常重要的一环。为了更加高效地进行数据采集,Flume作为一种流式数据采集工具得到了广泛的应用。其中,Flume的Channel模块是实现数据缓存和传输的核心模块之一。本文将介绍Flume中的Kafka Channel,讲解其数据采集流程。
65 0
|
3月前
|
消息中间件 数据采集 存储
大数据数据采集的数据采集(收集/聚合)的Flume之数据采集流程的Source的Kafka Source
在Flume中,Kafka Source是一种常见的Source类型。它可以从Kafka的Topic中采集数据,并将其转换成Flume事件进行处理和存储。本文将介绍Kafka Source的配置和数据采集流程。
71 0
|
8月前
|
SQL 消息中间件 分布式计算
如何查看spark与hadoop、kafka、Scala、flume、hive等兼容版本【适用于任何版本】
如何查看spark与hadoop、kafka、Scala、flume、hive等兼容版本【适用于任何版本】
440 0
如何查看spark与hadoop、kafka、Scala、flume、hive等兼容版本【适用于任何版本】
|
8月前
|
消息中间件 存储 Kafka
flume与kafka整合高可靠教程
flume与kafka整合高可靠教程
164 0
flume与kafka整合高可靠教程
|
9月前
|
消息中间件 监控 Kafka
Flume监控文件并将数据输入至Kafka
Flume监控文件并将数据输入至Kafka
|
9月前
|
消息中间件 分布式计算 Kafka
Spark Streaming实时流处理项目实战笔记——使用KafkaSInk将Flume收集到的数据输出到Kafka
Spark Streaming实时流处理项目实战笔记——使用KafkaSInk将Flume收集到的数据输出到Kafka
|
10月前
|
消息中间件 缓存 分布式计算
Flume 读取本地数据输出到 HDFS/Kafka
Flume 读取本地数据输出到 HDFS/Kafka
107 0
Flume 读取本地数据输出到 HDFS/Kafka
推荐文章
更多