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

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

2023-12-26 330
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 基于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
目录
相关文章
消息中间件 存储 传感器
447 0
阿里云开发者
|
11月前
|
消息中间件 SQL 关系型数据库
Flink CDC + Kafka 加速业务实时化
Flink CDC 是一种支持流批一体的分布式数据集成工具，通过 YAML 配置实现数据传输过程中的路由与转换操作。它已从单一数据源的 CDC 数据流发展为完整的数据同步解决方案，支持 MySQL、Kafka 等多种数据源和目标端（如 Delta Lake、Iceberg）。其核心功能包括多样化数据输入链路、Schema Evolution、Transform 和 Routing 模块，以及丰富的监控指标。相比传统 SQL 和 DataStream 作业，Flink CDC 提供更灵活的 Schema 变更控制和原始 binlog 同步能力。
阿里云开发者
1366 0
阿里云云原生
|
12月前
|
消息中间件 运维 Kafka
直播预告｜Kafka+Flink 双引擎实战：手把手带你搭建分布式实时分析平台！
直播预告｜Kafka+Flink 双引擎实战：手把手带你搭建分布式实时分析平台！
阿里云云原生
331 11
灵杰开发者
|
消息中间件 运维 Kafka
直播预告｜Kafka+Flink双引擎实战：手把手带你搭建分布式实时分析平台！
在数字化转型中，企业亟需从海量数据中快速提取价值并转化为业务增长动力。5月15日19:00-21:00，阿里云三位技术专家将讲解Kafka与Flink的强强联合方案，帮助企业零门槛构建分布式实时分析平台。此组合广泛应用于实时风控、用户行为追踪等场景，具备高吞吐、弹性扩缩容及亚秒级响应优势。直播适合初学者、开发者和数据工程师，参与还有机会领取定制好礼！扫描海报二维码或点击链接预约直播：[https://developer.aliyun.com/live/255088](https://developer.aliyun.com/live/255088)
灵杰开发者
695 35
直播预告｜Kafka+Flink双引擎实战：手把手带你搭建分布式实时分析平台！
灵杰开发者
|
9月前
|
存储 分布式计算 数据处理
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
阿里云实时计算Flink团队，全球领先的流计算引擎缔造者，支撑双11万亿级数据处理，推动Apache Flink技术发展。现招募Flink执行引擎、存储引擎、数据通道、平台管控及产品经理人才，地点覆盖北京、杭州、上海。技术深度参与开源核心，打造企业级实时计算解决方案，助力全球企业实现毫秒洞察。
灵杰开发者
784 0
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
灵杰开发者
|
消息中间件 关系型数据库 MySQL
Flink CDC 在阿里云实时计算Flink版的云上实践
本文整理自阿里云高级开发工程师阮航在Flink Forward Asia 2024的分享，重点介绍了Flink CDC与实时计算Flink的集成、CDC YAML的核心功能及应用场景。主要内容包括：Flink CDC的发展及其在流批数据处理中的作用；CDC YAML支持的同步链路、Transform和Route功能、丰富的监控指标；典型应用场景如整库同步、Binlog原始数据同步、分库分表同步等；并通过两个Demo展示了MySQL整库同步到Paimon和Binlog同步到Kafka的过程。最后，介绍了未来规划，如脏数据处理、数据限流及扩展数据源支持。
灵杰开发者
987 0
Flink CDC 在阿里云实时计算Flink版的云上实践
技术内容小助手
|
存储 关系型数据库 BI
实时计算UniFlow：Flink+Paimon构建流批一体实时湖仓
实时计算架构中，传统湖仓架构在数据流量管控和应用场景支持上表现良好，但在实际运营中常忽略细节，导致新问题。为解决这些问题，提出了流批一体的实时计算湖仓架构——UniFlow。该架构通过统一的流批计算引擎、存储格式（如Paimon）和Flink CDC工具，简化开发流程，降低成本，并确保数据一致性和实时性。UniFlow还引入了Flink Materialized Table，实现了声明式ETL，优化了调度和执行模式，使用户能灵活调整新鲜度与成本。最终，UniFlow不仅提高了开发和运维效率，还提供了更实时的数据支持，满足业务决策需求。
技术内容小助手
1250 7
zdl
|
消息中间件 运维 大数据
大数据实时计算产品的对比测评：实时计算Flink版 VS 自建Flink集群
本文介绍了实时计算Flink版与自建Flink集群的对比，涵盖部署成本、性能表现、易用性和企业级能力等方面。实时计算Flink版作为全托管服务，显著降低了运维成本，提供了强大的集成能力和弹性扩展，特别适合中小型团队和业务波动大的场景。文中还提出了改进建议，并探讨了与其他产品的联动可能性。总结指出，实时计算Flink版在简化运维、降低成本和提升易用性方面表现出色，是大数据实时计算的优选方案。
zdl
746 56
六月的雨在钉钉
|
SQL 运维 数据可视化
阿里云实时计算Flink版产品体验测评
阿里云实时计算Flink基于Apache Flink构建，提供一站式实时大数据分析平台，支持端到端亚秒级实时数据分析，适用于实时大屏、实时报表、实时ETL和风控监测等场景，具备高性价比、开发效率、运维管理和企业安全等优势。
六月的雨在钉钉
832 8

热门文章

最新文章

  • 1
    【最佳实践】实时计算 Flink 版在金融行业的实时数仓建设实践
  • 2
    Flink CDC 2.2 正式发布，新增四种数据源，支持动态加表，提供增量快照框架
  • 3
    Flink 靠什么征服饿了么工程师？
  • 4
    Apache Flink 漫谈系列(08) - SQL概览
  • 5
    带你读《Flink原理、实战与性能优化》之一：Apache Flink介绍
  • 6
    实时计算 Flink版操作报错合集之sqlserver mysql都用的胖包，sqlserver的成功了，mysql报这个错如何解决
  • 7
    在Flink的Standalone模式下，您可以使用`start-cluster.sh`命令启动Flink集群
  • 8
    Flink on Yarn编译的Parcels包到CDH集群实录
  • 9
    Flink sql 问题之主动使数据延时一段时间如何解决
  • 10
    【极数系列】Flink详细入门教程 & 知识体系 & 学习路线（01）
  • 1
    Linux下安装Kafka 3.9.1
    961
  • 2
    阿里云、Ververica、Confluent 与 LinkedIn 携手推进流式创新，共筑基于 Apache Flink Agents 的智能体 AI 未来
    1242
  • 3
    别再乱排查了！Kafka 消息积压、重复、丢失，根源基本都是 Rebalance！
    1425
  • 4
    AI 时代的数据通道：云消息队列 Kafka 的演进与实践
    682
  • 5
    爆款游戏背后：尚娱如何借助阿里云 Kafka Serverless 轻松驾驭“潮汐流量”？
    335
  • 6
    Apache Kafka 分布式流处理平台技术详解与实践指南
    755
  • 7
    嘉银科技基于阿里云 Kafka Serverless 提升业务弹性能力，节省成本超过 20%
    665
  • 8
    Spring实现Kafka重试Topic,真的太香了
    623
  • 9
    构建高效的实时数据管道：Flink、Kafka、CnosDB 的完美结合
    447
  • 10
    消息队列比较：Spring 微服务中的 Kafka 与 RabbitMQ
    552

    • 相关课程

    更多
  • Apache Flink 入门到实战 - Flink开源社区出品
  • 实时计算 Flink 版产品入门与实操
  • 开源 Flink 极速上手教程
  • 分布式消息系统 Kafka 快速入门
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门

    • 相关电子书

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

    • 相关实验场景

    更多
  • 基于Hologres+Flink搭建GitHub实时数据大屏
  • 基于Flink+Kafka实现订单评论实时分析
    • 下一篇
    阿里云网盘 Skill 上线：让 OpenClaw 的成果，手机一点就能发给客户