Kafka实战-Flume到Kafka-阿里云开发者社区

开发者社区> 潇湘隐者> 正文

Kafka实战-Flume到Kafka

简介:
+关注继续查看

1.概述

  前面给大家介绍了整个Kafka项目的开发流程,今天给大家分享Kafka如何获取数据源,即Kafka生产数据。下面是今天要分享的目录:

  • 数据来源
  • Flume到Kafka
  • 数据源加载
  • 预览

  下面开始今天的分享内容。

2.数据来源

  Kafka生产的数据,是由Flume的Sink提供的,这里我们需要用到Flume集群,通过Flume集群将Agent的日志收集分发到 Kafka(供实时计算处理)和HDFS(离线计算处理)。关于Flume集群的Agent部署,这里就不多做赘述了,不清楚的同学可以参考《高可用Hadoop平台-Flume NG实战图解篇》一文中的介绍,下面给大家介绍数据来源的流程图,如下图所示:

  这里,我们使用Flume作为日志收集系统,将收集到的数据输送到Kafka中间件,以供Storm去实时消费计算,整个流程从各个Web节点 上,通过Flume的Agent代理收集日志,然后汇总到Flume集群,在由Flume的Sink将日志输送到Kafka集群,完成数据的生产流程。

3.Flume到Kafka

  从图,我们已经清楚了数据生产的流程,下面我们来看看如何实现Flume到Kafka的输送过程,下面我用一个简要的图来说明,如下图所示:

  这个表达了从Flume到Kafka的输送工程,下面我们来看看如何实现这部分。

  首先,在我们完成这部分流程时,需要我们将Flume集群和Kafka集群都部署完成,在完成部署相关集群后,我们来配置Flume的Sink数据流向,配置信息如下所示:

  • 首先是配置spooldir方式,内容如下所示:

producer.sources.s.type = spooldir
producer.sources.s.spoolDir = /home/hadoop/dir/logdfs

  • 当然,Flume的数据发送方类型也是多种类型的,有:Console、Text、HDFS、RPC等,这里我们系统所使用的是Kafka中间件来接收,配置内容如下所示:


producer.sinks.r.type = org.apache.flume.plugins.KafkaSink
producer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092
producer.sinks.r.partition.key=0
producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition
producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
producer.sinks.r.request.required.acks=0
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
producer.sinks.r.custom.topic.name=test

  这样,我们就在Flume的Sink端配置好了数据流向接受方。

4.数据加载

  在完成配置后,接下来我们开始加载数据,首先我们在Flume的spooldir端生产日志,以供Flume去收集这些日志。然后,我们通过Kafka的KafkaOffsetMonitor监控工具,去监控数据生产的情况,下面我们开始加载。

  • 启动ZK集群,内容如下所示:




zkServer.sh start

  注意:分别在ZK的节点上启动。

  • 启动Kafka集群




kafka-server-start.sh config/server.properties &

  在其他的Kafka节点输入同样的命令,完成启动。

  • 启动Kafka监控工具


java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \
 com.quantifind.kafka.offsetapp.OffsetGetterWeb \
 --zk dn1:2181,dn2:2181,dn3:2181 \
 --port 8089 \
 --refresh 10.seconds \
 --retain 1.days

  启动Flume集群





flume-ng agent -n producer -c conf -f flume-kafka-sink.properties -Dflume.root.logger=ERROR,console

  然后,我在/home/hadoop/dir/logdfs目录下上传log日志,这里我只抽取了一少部分日志进行上传,如下图所示,表示日志上传成功。

5.预览

  下面,我们通过Kafka的监控工具,来预览我们上传的日志记录,有没有在Kafka中产生消息数据,如下所示:

  • 启动Kafka集群,为生产消息截图预览

  • 通过Flume上传日志,在Kafka中产生消息数据

6.总结

  本篇文章给大家讲述了Kafka的消息产生流程,后续会在Kafka实战系列中为大家讲述Kafka的消息消费流程等一整套流程,这里只是为后续的Kafka实战编码打下一个基础,让大家先对Kafka的消息生产有个整体的认识。

7.结束语

  这篇博客就和大家分享到这里,如果大家在研究学习的过程当中有什么问题,可以加群进行讨论或发送邮件给我,我会尽我所能为您解答,与君共勉!

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云消息队列Kafka实战经验分享
消息队列Kafka版是阿里云提供的分布式、高吞吐、可扩展的消息队列服务。消息队列Kafka版广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域,已成为大数据生态中不可或缺的一部分。本次直播将为您深入分享消息队列Kafka的实战经验。
6895 0
带你读《Flutter技术入门与实战》之三:Dart语言简述
本书的定位适合小白程序员,入门加实战,既有基础知识,又有丰富示例,包括详细的操作步骤,实操性强。由于Flutter大量使用组件,所以对组件的讲解很详细,包括基本概念、属性及代码示例。每个组件都配有小例子,力求精简,还提供了配套网站提供完整代码,复制完整代码就可以立即看到效果。在轻松掌握基础知识的同时快速进入实战。
983 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
4615 0
Knative 实战:基于 Kafka 实现消息推送
作者 | 元毅 阿里云智能事业群高级开发工程师 导读:当前在 Knative 中已经提供了对 Kafka 事件源的支持,那么如何基于 Kafka 实现消息推送呢?本文作者将以阿里云 Kafka 产品为例,给大家解锁这一新的姿势。
1670 0
《Apache Kafka实战》| 每日读本书
基于Apache Kafka 1.0.0版本进行介绍,Kafka Contributor执笔。包括Kafka基本概念与特性,以及Kafka的部署、开发、运营、监控、调试、优化以及重要组件的设计原理,并给出了翔实的案例。每日搜罗最具权威专业书籍,更多图书请关注“每日读本书”。
1339 0
Kafka实战(四) -Kafka门派知多少
Kafka不是一个单纯的消息引擎系统,而是能够实现精确一次(Exactly-once)处理语义的实时流处理平台 Storm/Spark Streaming/Flink,在大规模流处理领域主流 Kafka经过这么长时间不断的迭代,现在已经能够稍稍比肩这些框架 Kafka社区对于这些框架心存敬意 目前国内鲜有大厂将Kafka用于流处理的尴尬境地,毕竟Kafka是从消息引擎“半路出家”转型成流处理平台的,它在流处理方面的表现还需要经过时间的检验。
2006 0
+关注
潇湘隐者
网名潇湘隐者/潇湘剑客、英文名Kerry,兴趣广泛,广泛涉猎,个性随意,不善言辞。执意做一名会写代码的DBA,混迹于IT行业
777
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
文娱运维技术
立即下载
《SaaS模式云原生数据仓库应用场景实践》
立即下载
《看见新力量:二》电子书
立即下载