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.结束语

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

相关文章
|
监控 安全 Linux
【Elasticsearch专栏 14】深入探索:Elasticsearch使用Logstash的日期过滤器删除旧数据
使用Logstash的日期过滤器可以有效删除Elasticsearch中的旧数据,释放存储空间并提高集群性能。通过配置Logstash,可以指定索引模式、筛选时间戳早于特定阈值的文档,并在输出阶段删除这些旧数据。执行配置时,需确保Logstash与Elasticsearch连接正常,并监控日志以确保操作安全。定期执行此操作可确保旧数据不会过多积累。总之,Logstash的日期过滤器提供了一种简单而高效的方法,帮助管理和优化Elasticsearch中的数据。
517 0
|
9月前
|
数据采集 监控 算法
Playwright错误处理与重试机制实现
本文详解如何为Playwright脚本构建健壮的错误处理与重试机制,涵盖智能等待、异常捕获、重试策略、错误分类、上下文恢复及监控日志,提升自动化测试与爬虫的稳定性与可靠性。
|
测试技术
【sgLazyCascader】自定义组件:基于el-cascader的懒加载级联菜单,支持异步加载子级菜单
【sgLazyCascader】自定义组件:基于el-cascader的懒加载级联菜单,支持异步加载子级菜单
|
人工智能 自然语言处理 程序员
用通义灵码@workspace和@terminal做项目代码熟悉和新需求实现的体验
来和大家分享一下,我用通义灵码@workspace和@terminal做项目代码熟悉和新需求实现的体验。
952 2
|
搜索推荐 API 开发者
Django框架和Flask框架的适用场景分别是什么?
总体而言,Django 更适合需要全面功能和大规模开发的场景,而 Flask 则更适合灵活性要求高、小型项目或特定需求的开发。当然,具体的选择还应根据项目的具体情况、团队的技术能力和偏好等因素来综合考虑。在实际应用中,开发者可以根据项目的特点和需求,灵活选择使用这两个框架,或者结合它们的优势来构建更强大的 Web 应用程序。
1061 163
|
运维 监控 网络协议
如何设计一个高性能内网穿透服务
本文深入解析了基于Go语言开发的高性能内网穿透工具IntraLink,涵盖其技术架构、核心实现、协议处理、性能优化与安全机制,展示了如何构建高效、稳定、安全的内网穿透服务。
491 1
|
Ubuntu
ubuntu 扫描局域网ip
ubuntu 扫描局域网ip ifconfig
984 0
ubuntu 扫描局域网ip
|
Java 数据安全/隐私保护 Python
Python案例分享:如何实现文件的解压缩
Python案例分享:如何实现文件的解压缩
660 8
|
机器学习/深度学习 人工智能 自然语言处理
【机器学习】python之人工智能应用篇--代码生成技术
代码生成技术是人工智能与软件工程交叉领域的一项重要技术,它利用机器学习、自然语言处理和其他AI算法自动编写或辅助编写计算机程序代码。这一技术旨在提高编程效率、降低错误率,并帮助非专业开发者快速实现功能。以下是代码生成技术的概述及其典型应用场景。
607 6
|
存储 运维 Shell
Ansible自动化运维工具安装和基本使用
Ansible 是一款无代理的IT自动化工具,通过SSH连接目标主机执行配置管理、应用部署和云端管理任务。它使用YAML编写的Playbook定义任务,核心组件包括Playbook、模块、主机清单、变量等。Ansible的优势在于易用、功能强大、无须在目标主机安装额外软件,并且开源。安装过程涉及配置网络源、yum安装和SSH密钥设置。通过定义主机清单和使用模块进行通信测试,确保连接成功。
851 2
Ansible自动化运维工具安装和基本使用

热门文章

最新文章