AI 助理
文档备案控制台
开发者社区 微服务 文章 正文

基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(二)数据源

2023-12-26 387
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(二)数据源

04：数据源

  • 目标：了解数据源的格式及实现模拟数据的生成
  • 路径
  • step1：数据格式
  • step2：数据生成
  • 实施
  • 数据格式
消息时间 发件人昵称 发件人账号 发件人性别 发件人IP 发件人系统 发件人手机型号 发件人网络制式 发件人GPS 收件人昵称 收件人IP 收件人账号 收件人系统 收件人手机型号 收件人网络制式 收件人GPS 收件人性别 消息类型 双方距离 消息
msg_time sender_nickyname sender_account sender_sex sender_ip sender_os sender_phone_type sender_network sender_gps receiver_nickyname receiver_ip receiver_account receiver_os receiver_phone_type receiver_network receiver_gps receiver_sex msg_type distance message
2020/05/08 15:11:33 古博易 14747877194 男 48.147.134.255 Android 8.0 小米 Redmi K30 4G 94.704577,36.247553 莱优 97.61.25.52 17832829395 IOS 10.0 Apple iPhone 10 4G 84.034145,41.423804 女 TEXT 77.82KM 天涯海角惆怅渡，牛郎织女隔天河。佛祖座前长顿首，只求共度一百年。

  • 数据生成
  • 创建原始文件目录
mkdir /export/data/momo_init
cd /export/data/momo_init
rz
  • 创建模拟数据目录
mkdir /export/data/momo_data
  • 运行程序生成数据
  • 语法
java -jar /export/data/momo_init/MoMo_DataGen.jar 原始数据路径 模拟数据路径 随机产生数据间隔ms时间
  • 测试：每500ms生成一条数据
java -jar /export/data/momo_init/MoMo_DataGen.jar \
/export/data/momo_init/MoMo_Data.xlsx \
/export/data/momo_data/ \
500
  • 结果：生成模拟数据文件MOMO_DATA.dat，并且每条数据中字段分隔符为\001

  • 小结
  • 了解数据源的格式及实现模拟数据的生成

05：技术架构及技术选型

  • 目标：掌握实时案例的技术架构及技术选型
  • 路径
  • step1：需求分析
  • step2：技术选型
  • step3：技术架构
  • 实施
  • 需求分析
  • 离线存储计算
  • 提供离线T + 1的统计分析
  • 提供离线数据的即时查询
  • 实时存储计算
  • 提供实时统计分析
  • 技术选型
  • 离线
  • 数据采集：Flume
  • 离线存储：Hbase
  • 离线分析：Hive：复杂计算
  • 即时查询：Phoenix：高效查询
  • 实时
  • 数据采集：Flume
  • 实时存储：Kafka
  • 实时计算：Flink
  • 实时应用：MySQL + FineBI 或者 Redis + JavaWeb可视化
  • 技术架构
  • 为什么不直接将Flume的数据给Hbase，而统一的给了Kafka，再由Kafka到Hbase？
  • 避免高并发写导致机器负载过高、实现架构解耦、实现异步高效
  • 保证数据一致性
  • 小结
  • 掌握实时案例的技术架构及技术选型

06：Flume的回顾及安装

  • 目标：回顾Flume基本使用及实现Flume的安装测试
  • 路径
  • step1：Flume回顾
  • step2：Flume的安装
  • step3：Flume的测试
  • 实施
  • Flume的回顾
  • 功能：实时对文件或者网络端口进行数据流监听采集
  • 场景：文件实时采集
  • 开发
  • step1：先开发一个配置文件：properties【K=V】
  • step2：运行这个文件即可
  • 组成
  • Agent：一个Agent就是一个Flume程序
  • Source：负责监听数据源，将数据源的动态数据变成每一条Event数据，将Event数据流放入Channel
  • Channel：负责临时存储Source发送过来的数据，供Sink来取数据
  • Sink：负责从Channel拉取数据写入目标地
  • Event：代表一条数据对象
  • head：Map集合[KV]
  • body：byte[]
  • Flume的安装
  • 上传安装包
cd /export/software/
rz
  • 解压安装
tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /export/server/
cd /export/server
mv apache-flume-1.9.0-bin flume-1.9.0-bin
  • 修改配置
#集成HDFS，拷贝HDFS配置文件
cd /export/server/flume-1.9.0-bin
cp /export/server/hadoop/etc/hadoop/core-site.xml  ./conf/
#修改Flume环境变量
cd /export/server/flume-1.9.0-bin/conf/
mv flume-env.sh.template flume-env.sh
vim flume-env.sh 
#修改22行
export JAVA_HOME=/export/server/jdk1.8.0_65
#修改34行
export HADOOP_HOME=/export/server/hadoop-3.3.0
  • 删除Flume自带的guava包，替换成Hadoop的
cd /export/server/flume-1.9.0-bin 
rm -rf lib/guava-11.0.2.jar
cp /export/server/hadoop/share/hadoop/common/lib/guava-27.0-jre.jar lib/
  • 创建目录
cd /export/server/flume-1.9.0-bin
#程序配置文件存储目录
mkdir usercase
#Taildir元数据存储目录
mkdir position
  • Flume的测试
  • 需求：采集聊天数据，写入HDFS
  • 分析
  • Source：taildir：动态监听多个文件实现实时数据采集
  • Channel：mem：将数据缓存在内存
  • Sink：hdfs
  • 开发
vim /export/server/flume-1.9.0-bin/usercase/momo_mem_hdfs.properties
# define a1
a1.sources = s1 
a1.channels = c1
a1.sinks = k1
#define s1
a1.sources.s1.type = TAILDIR
#指定一个元数据记录文件
a1.sources.s1.positionFile = /export/server/flume-1.9.0-bin/position/taildir_momo_hdfs.json
#将所有需要监控的数据源变成一个组
a1.sources.s1.filegroups = f1
#指定了f1是谁：监控目录下所有文件
a1.sources.s1.filegroups.f1 = /export/data/momo_data/.*
#指定f1采集到的数据的header中包含一个KV对
a1.sources.s1.headers.f1.type = momo
a1.sources.s1.fileHeader = true
#define c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000
#define k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/momo/test/daystr=%Y-%m-%d
a1.sinks.k1.hdfs.fileType = DataStream
#指定按照时间生成文件，一般关闭
a1.sinks.k1.hdfs.rollInterval = 0
#指定文件大小生成文件，一般120 ~ 125M对应的字节数
a1.sinks.k1.hdfs.rollSize = 102400
#指定event个数生成文件，一般关闭
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.filePrefix = momo
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#bound
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1
  • 启动HDFS
start-dfs.sh
  • 运行Flume
cd /export/server/flume-1.9.0-bin
bin/flume-ng agent -c conf/ -n a1 -f usercase/momo_mem_hdfs.properties -Dflume.root.logger=INFO,console
  • 运行模拟数据
java -jar /export/data/momo_init/MoMo_DataGen.jar \
/export/data/momo_init/MoMo_Data.xlsx \
/export/data/momo_data/ \
100
  • 查看结果

  • 小结
  • 回顾Flume基本使用及实现Flume的安装测试

07：Flume采集程序开发

  • 目标：实现案例Flume采集程序的开发
  • 路径
  • step1：需求分析
  • step2：程序开发
  • step3：测试实现
  • 实施
  • 需求分析
  • 需求：采集聊天数据，实时写入Kafka
  • Source：taildir
  • Channel：mem
  • Sink：Kafka sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers = node1:9092,node2:9092,node3:9092
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy
  • 程序开发
vim /export/server/flume-1.9.0-bin/usercase/momo_mem_kafka.properties
# define a1
a1.sources = s1 
a1.channels = c1
a1.sinks = k1
#define s1
a1.sources.s1.type = TAILDIR
#指定一个元数据记录文件
a1.sources.s1.positionFile = /export/server/flume-1.9.0-bin/position/taildir_momo_kafka.json
#将所有需要监控的数据源变成一个组
a1.sources.s1.filegroups = f1
#指定了f1是谁：监控目录下所有文件
a1.sources.s1.filegroups.f1 = /export/data/momo_data/.*
#指定f1采集到的数据的header中包含一个KV对
a1.sources.s1.headers.f1.type = momo
a1.sources.s1.fileHeader = true
#define c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000
#define k1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = MOMO_MSG
a1.sinks.k1.kafka.bootstrap.servers = node1:9092,node2:9092,node3:9092
a1.sinks.k1.kafka.flumeBatchSize = 10
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 100
#bound
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1
  • 测试实现
  • 启动Kafka
start-zk-all.sh
start-kafka.sh
  • 创建Topic
kafka-topics.sh --create --topic MOMO_MSG  --partitions 3 --replication-factor 2 --bootstrap-server node1:9092,node2:9092,node3:9092

注意：Kafka2.11版本用–zookeeper 替代

kafka-topics.sh --create --topic MOMO_MSG --partitions 3 --replication-factor 2 --zookeeper node01:9092

  • 列举
kafka-topics.sh --list --bootstrap-server node1:9092,node2:9092,node3:9092
  • 启动消费者
kafka-console-consumer.sh --topic MOMO_MSG --bootstrap-server node1:9092,node2:9092,node3:9092
  • 启动Flume程序
cd /export/server/flume-1.9.0-bin
bin/flume-ng agent -c conf/ -n a1 -f usercase/momo_mem_kafka.properties -Dflume.root.logger=INFO,console
  • 启动模拟数据
java -jar /export/data/momo_init/MoMo_DataGen.jar \
/export/data/momo_init/MoMo_Data.xlsx \
/export/data/momo_data/ \
50
  • 观察结果

  • 小结
  • 实现案例Flume采集程序的开发


文章标签：
云消息队列 Kafka 版
云数据库 HBase 版
实时计算 Flink版
消息中间件
Kafka
存储
数据采集
分布式数据库
关键词：
云消息队列 Kafka 版flink
实时计算 Flink版kafka
云数据库 HBase 版案例
实时计算 Flink版案例
云消息队列 Kafka 版案例
Maynor
目录
相关文章
消息中间件 存储 传感器
539 0
阿里云开发者
|
消息中间件 SQL 关系型数据库
Flink CDC + Kafka 加速业务实时化
Flink CDC 是一种支持流批一体的分布式数据集成工具，通过 YAML 配置实现数据传输过程中的路由与转换操作。它已从单一数据源的 CDC 数据流发展为完整的数据同步解决方案，支持 MySQL、Kafka 等多种数据源和目标端（如 Delta Lake、Iceberg）。其核心功能包括多样化数据输入链路、Schema Evolution、Transform 和 Routing 模块，以及丰富的监控指标。相比传统 SQL 和 DataStream 作业，Flink CDC 提供更灵活的 Schema 变更控制和原始 binlog 同步能力。
阿里云开发者
1539 0
阿里云云原生
|
消息中间件 运维 Kafka
直播预告｜Kafka+Flink 双引擎实战：手把手带你搭建分布式实时分析平台！
直播预告｜Kafka+Flink 双引擎实战：手把手带你搭建分布式实时分析平台！
阿里云云原生
373 11
灵杰开发者
|
消息中间件 运维 Kafka
直播预告｜Kafka+Flink双引擎实战：手把手带你搭建分布式实时分析平台！
在数字化转型中，企业亟需从海量数据中快速提取价值并转化为业务增长动力。5月15日19:00-21:00，阿里云三位技术专家将讲解Kafka与Flink的强强联合方案，帮助企业零门槛构建分布式实时分析平台。此组合广泛应用于实时风控、用户行为追踪等场景，具备高吞吐、弹性扩缩容及亚秒级响应优势。直播适合初学者、开发者和数据工程师，参与还有机会领取定制好礼！扫描海报二维码或点击链接预约直播：[https://developer.aliyun.com/live/255088](https://developer.aliyun.com/live/255088)
灵杰开发者
758 35
直播预告｜Kafka+Flink双引擎实战：手把手带你搭建分布式实时分析平台！
技术小达人
|
消息中间件 关系型数据库 MySQL
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
技术小达人
1548 0
赵渝强老师
|
9月前
|
数据采集 缓存 大数据
【赵渝强老师】大数据日志采集引擎Flume
Apache Flume 是一个分布式、可靠的数据采集系统，支持从多种数据源收集日志信息，并传输至指定目的地。其核心架构由Source、Channel、Sink三组件构成，通过Event封装数据，保障高效与可靠传输。
赵渝强老师
485 1
武子康
|
SQL 分布式计算 Hadoop
Hadoop-19 Flume Agent批量采集数据到HDFS集群 监听Hive的日志 操作则把记录写入到HDFS 方便后续分析
Hadoop-19 Flume Agent批量采集数据到HDFS集群 监听Hive的日志 操作则把记录写入到HDFS 方便后续分析
武子康
388 2
武子康
|
存储 数据采集 分布式计算
Hadoop-17 Flume 介绍与环境配置 实机云服务器测试 分布式日志信息收集 海量数据 实时采集引擎 Source Channel Sink 串行复制负载均衡
Hadoop-17 Flume 介绍与环境配置 实机云服务器测试 分布式日志信息收集 海量数据 实时采集引擎 Source Channel Sink 串行复制负载均衡
武子康
376 1
土木林森
|
存储 分布式计算 大数据
【Flume的大数据之旅】探索Flume如何成为大数据分析的得力助手，从日志收集到实时处理一网打尽！
【8月更文挑战第24天】Apache Flume是一款高效可靠的数据收集系统，专为Hadoop环境设计。它能在数据产生端与分析/存储端间搭建桥梁，适用于日志收集、数据集成、实时处理及数据备份等多种场景。通过监控不同来源的日志文件并将数据标准化后传输至Hadoop等平台，Flume支持了性能监控、数据分析等多种需求。此外，它还能与Apache Storm或Flink等实时处理框架集成，实现数据的即时分析。下面展示了一个简单的Flume配置示例，说明如何将日志数据导入HDFS进行存储。总之，Flume凭借其灵活性和强大的集成能力，在大数据处理流程中占据了重要地位。
土木林森
525 3
土木林森
|
存储 数据采集 数据处理
【Flume拓扑揭秘】掌握Flume的四大常用结构，构建强大的日志收集系统！
【8月更文挑战第24天】Apache Flume是一个强大的工具，专为大规模日志数据的收集、聚合及传输设计。其核心架构包括源（Source）、通道（Channel）与接收器（Sink）。Flume支持多样化的拓扑结构以适应不同需求，包括单层、扇入（Fan-in）、扇出（Fan-out）及复杂多层拓扑。单层拓扑简单直观，适用于单一数据流场景；扇入结构集中处理多源头数据；扇出结构则实现数据多目的地分发；复杂多层拓扑提供高度灵活性，适合多层次数据处理。通过灵活配置，Flume能够高效构建各种规模的数据收集系统。
土木林森
546 0

热门文章

最新文章

  • 1
    HBase进化之从NoSQL到NewSQL，凤凰涅槃成就Phoenix
  • 2
    如何降低90%Java垃圾回收时间？以阿里HBase的GC优化实践为例
  • 3
    深入解读HBase2.0新功能之高可用读Region Replica
  • 4
    双研究员带你了解数据库技术现状，及阿里云为什么要推出HBase
  • 5
    【干货合集】NoSQL技术体系深度解读系列（三）：HBase，海量数据存储、超高并发量场景下的NoSQL利器
  • 6
    hadoop2.6完全分布式安装HBase1.1
  • 7
    HBase指南 | HBase 2.0之修复工具HBCK2运维指南
  • 8
    HBase写性能优化
  • 9
    【HBase】（七）Hbase 常用API（增删改查）
  • 10
    Hbase操作table常见方法示例
  • 1
    【赵渝强老师】大数据日志采集引擎Flume
    485
  • 2
    踏上大数据第一步：flume
    522
  • 3
    Flume+Hadoop：打造你的大数据处理流水线
    1016
  • 4
    Hadoop-20 Flume 采集数据双写至本地+HDFS中 监控目录变化 3个Agent MemoryChannel Source对比
    312
  • 5
    Hadoop-19 Flume Agent批量采集数据到HDFS集群 监听Hive的日志 操作则把记录写入到HDFS 方便后续分析
    388
  • 6
    Hadoop-18 Flume HelloWorld 第一个Flume尝试！编写conf实现Source+Channel+Sink 控制台查看收集到的数据 流式收集
    355
  • 7
    Hadoop-17 Flume 介绍与环境配置 实机云服务器测试 分布式日志信息收集 海量数据 实时采集引擎 Source Channel Sink 串行复制负载均衡
    376
  • 8
    Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
    288
  • 9
    Flume核心组件大揭秘：Agent、Source、Channel、Sink，一文掌握数据采集精髓！
    1812
  • 10
    【Flume的大数据之旅】探索Flume如何成为大数据分析的得力助手，从日志收集到实时处理一网打尽！
    525

    • 相关课程

    更多
  • 大数据Flink实时旅游平台环境篇 2020版
  • Apache Flink 入门到实战 - Flink开源社区出品
  • 实时计算 Flink 版产品入门与实操
  • 开源 Flink 极速上手教程
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门

    • 相关电子书

    更多
  • Flink峰会 - 李佳林
  • Flink峰会 - 徐榜江
  • Flink CDC Meetup PPT - 龚中强

    • 相关实验场景

    更多
  • 基于Hologres+Flink搭建GitHub实时数据大屏
  • 基于Flink+Kafka实现订单评论实时分析
    • 下一篇
    一条命令迁移，帮你实现 OpenClaw 与 Hermes Agent 记忆互通！