案例1:syslog-memory-kafka

将flume采集到的数据落地到kafka上,即sink是kafka(生产者身份)

vim syslog-mem-kafka.conf
# 命名个组件
a1.sources = r1
a1.sinks = k1
a1.channels = c1#source属性
a1.sources.r1.type = syslogtcp
a1.sources.r1.host=mypc01
a1.sources.r1.port=10086# 描述channel属性
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# 描述sink属性
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers = mypc01:9092,mypc:9092,mypc03:9092
# 主题必须提前存在
a1.sinks.k1.kafka.topic = pet
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1# 关联source和sink到channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动flume

#bin/bash
/usr/local/flume/bin/flume-ng agent -c /usr/local/flume/conf \
-f /usr/local/flume/flumeconf/syslog-mem-kafka.conf \
-n a1 -Dflume.root.logger=INFO,console -Dflume.monitoring.type=http -Dflume.monitoring.port=31002

先启动消费者准备接受消息

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet

测试

echo "aaaaa" | nc mypc01 10086

案例2 kafka-memory-hdfs

kafka的source类型从kafka集群读取数据,就是消费者身份,将数据封装成event落地到hdfs

vim kafka-mem-kafka.conf
# 命名个组件
a1.sources = r1
a1.sinks = k1
a1.channels = c1#source属性
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.kafka.bootstrap.servers = mypc01:9092,mypc02:9092,mypc03:9092
a1.sources.r1.kafka.consumer.group.id=g1
a1.sources.r1.kafka.topics=pet# 描述channel属性
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# 描述sink属性
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=hdfs://mypc01:8020/kafka/pet/%Y%m%d
a1.sinks.k1.hdfs.filePrefix=FlumeData
a1.sinks.k1.hdfs.fileSuffix = .kafka
a1.sinks.k1.hdfs.rollSize=102400
a1.sinks.k1.hdfs.rollCount  = 0
#单位为s
b1001.sinks.k1.hdfs.rollInterval=60
b1001.sinks.k1.hdfs.useLocalTimeStamp = true# 关联source和sink到channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动flume

#bin/bash
/usr/local/flume/bin/flume-ng agent -c /usr/local/flume/conf \
-f /usr/local/flume/flumeconf/kafka-mem-hdfs.conf \
-n a1 -Dflume.root.logger=INFO,console -Dflume.monitoring.type=http -Dflume.monitoring.port=31002

启动生产者,使用生产者发送消息

kafka-console-producer.sh \
--broker-list mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet

之后就可以在hdfs上看到生成的文件了.

总结

kafka可以作为source,也可以作为sink

flume与kafka的整合相关推荐

大数据集群搭建（12）——Flume和Kafka的整合
Flume和Kafka的整合 1.配置flume,在flume的conf目录下新建文件(flume_kafka.conf)并配置. ################################# ...
Flume与Kafka整合案例详解
环境配置名称版本下载地址 Centos 7.0 64x 百度 Zookeeper 3.4.5 Flume 1.6.0 Kafka 2.1.0 flume笔记直接贴配置文件 [roo ...
大数据———Flume与Kafka整合
环境配置名称版本下载地址 Centos 7.0 64x 百度 Flume 1.8.0 http://flume.apache.org/download.html Kafka 2.11 http: ...
Flume+Kafka+SparkStreaming整合
目录 1.Flume介绍.2 1.1 Flume数据源以及输出方式.2 1.2 Flume的核心概念.2 1.3 Flume结构.2 1.4 Flume安装测试.3 1.5 启动flume4 2.Ka ...
整合Flume和Kafka完成实时数据采集
需要注意:参考的网站要与你的kafka的版本一致,因为里面的字段会不一致例如:http://flume.apache.org/releases/content/1.6.0/FlumeUserGuid ...
flume消费kafka数据太慢_kafka补充01
为什么高吞吐? •写数据 –1.页缓存技术 •kafka写出数据时先将数据写到操作系统的pageCache上,由操作系统自己决定什么时候将数据写到磁盘上 –2.磁盘顺序写 •磁盘顺序写的性能会比随机写 ...
[大数据] 搜索日志数据采集系统 flume+hbase+kafka架构（数据搜狗实验室）
1 采集规划说明: D1 日志所在服务器1 -bigdata02.com D2 日志所在服务器2 -bigdata03.com A flume2 - bigdata02.com 日志收集 C flu ...
大数据流处理：Flume、Kafka和NiFi对比
在构建大数据流水线时,我们需要考虑处理数据的数量,种类和速度,这些数据通常出现在Hadoop生态系统的入口.在决定采用哪种工具来满足我们的要求时,都会考虑到可扩展性.可靠性.适应性.开发时间方面的成本 ...
Kafka实战－Flume到Kafka
1.概述前面给大家介绍了整个Kafka项目的开发流程,今天给大家分享Kafka如何获取数据源,即Kafka生产数据.下面是今天要分享的目录: 数据来源 Flume到Kafka 数据源加载预览下面 ...

flume与kafka的整合

案例1:syslog-memory-kafka

案例2 kafka-memory-hdfs

总结

flume与kafka的整合相关推荐

最新文章

热门文章