Flume+Kafka+SparkStreaming的整合是怎么样的

63次阅读
没有评论

共计 2682 个字符,预计需要花费 7 分钟才能阅读完成。

本篇文章给大家分享的是有关 Flume+Kafka+SparkStreaming 的整合是怎么样的,丸趣 TV 小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获,话不多说,跟着丸趣 TV 小编一起来看看吧。

1. 架构

第一步,Flume 和 Kakfa 对接,Flume 抓取日志,写到 Kafka 中

第二部,Spark Streaming 读取 Kafka 中的数据,进行实时分析

下面首先使用 Kakfa 自带的消息处理(脚本)来获取消息,走通 Flume 和 Kafka 的对接

2. 安装 flume,kafka

flume install: http://my.oschina.net/u/192561/blog/692225

kafka install: http://my.oschina.net/u/192561/blog/692357

3.Flume 和 Kafka 整合

3.1 两者整合优势

Flume 更倾向于数据传输本身,Kakfa 是典型的消息中间件用于解耦生产者消费者。

具体架构上,Agent 并没把数据直接发送到 Kafka,在 Kafka 前面有层由 Flume 构成的 forward。这样做有两个原因:

Kafka 的 API 对非 JVM 系的语言支持很不友好,forward 对外提供更加通用的 HTTP 接口。forward 层可以做路由、Kafka topic 和 Kafkapartition key 等逻辑,进一步减少 Agent 端的逻辑。

数据有数据源到 flume 再到 Kafka 时,数据一方面可以同步到 HDFS 做离线计算,另一方面可以做实时计算。本文实时计算采用 SparkStreaming 做测试。
 

3.2 Flume 和 Kafka 整合安装

1. 下载 Flume 和 Kafka 集成的插件,下载地址:

https://github.com/beyondj2ee/flumeng-kafka- plugin

将 package 目录中的 flumeng-kafka-plugin.jar 拷贝到 Flume 安装目录的 lib 目录下

2. 将 Kakfa 安装目录 libs 目录下的如下 jar 包拷贝到 Flume 安装目录的 lib 目录下

kafka_2.11-0.10.0.0.jar

scala-library-2.11.8.jar

metrics-core-2.2.0.jar

提取插件中的 flume-conf.properties 文件:修改如下:flume 源采用 exec

producer.sources.s.type = exec
producer.sources.s.command=tail -F -n+1 /home/eric/bigdata/kafka-logs/a.log
producer.sources.s.channels = c1

修改 producer 代理的 topic 为 HappyBirthDayToAnYuan

将配置放到 apache-flume-1.6.0-bin/conf/producer.conf 中

  完整 producer.conf: 

#agentsectionproducer.sources= s1producer.channels= c1producer.sinks= k1#配置数据源 producer.sources.s1.type=exec#配置需要监控的日志输出文件或目录 producer.sources.s1.command=tail -F -n+1 /home/eric/bigdata/kafka-logs/a.log#配置数据通道 producer.channels.c1.type=memoryproducer.channels.c1.capacity=10000producer.channels.c1.transactionCapacity=100# 配置数据源输出 #设置 Kafka 接收器, 此处最坑, 注意版本, 此处为 Flume 1.6.0 的输出槽类型 producer.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink#设置 Kafka 的 broker 地址和端口号 producer.sinks.k1.brokerList=localhost:9092#设置 Kafka 的 Topicproducer.sinks.k1.topic=HappyBirthDayToAnYuan# 设置序列化方式 producer.sinks.k1.serializer. >

3.3 启动 kafka flume 相关服务

启动 ZK  bin/zookeeper-server-start.sh config/zookeeper.properties

启动 Kafka 服务 bin/kafka-server-start.sh config/server.properties

创建主题

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic HappyBirthDayToAnYuan

查看主题

bin/kafka-topics.sh --list --zookeeper localhost:2181

查看主题详情

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic HappyBirthDayToAnYuan

删除主题

bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic test

创建消费者

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

启动 flume

bin/flume-ng agent -n producer -c conf -f conf/producer.conf -Dflume.root.logger=INFO,console

向 flume 发送数据:

echo yuhai a.log

kafka 消费数据:

注意: 当前文件内容删除, 服务器重启, 主题需重新创建, 但是消费内容有落地文件, 当前消费内容不消失.

以上就是 Flume+Kafka+SparkStreaming 的整合是怎么样的,丸趣 TV 小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注丸趣 TV 行业资讯频道。

正文完
 
丸趣
版权声明:本站原创文章,由 丸趣 2023-08-16发表,共计2682字。
转载说明:除特殊说明外本站除技术相关以外文章皆由网络搜集发布,转载请注明出处。
评论(没有评论)