阿里云大数据利器之-使用sql实现流计算做实时展现业务( flume故障转移版 )

本文涉及的产品
RDS AI 助手,专业版
RDS MySQL DuckDB 分析主实例,基础系列 4核8GB
RDS MySQL DuckDB 分析主实例,集群系列 4核8GB
简介: 实时业务处理的需求越来越多,也有各种处理方案,比如storm,spark等都可以。那以数据流的方向可以总结成数据源-数据搜集-缓存队列-实时处理计算-数据展现。本文就用阿里云产品简单实现了一个实时处理的方案。

实时业务处理的需求越来越多,也有各种处理方案,比如storm,spark等都可以。那以数据流的方向可以总结成数据源-数据搜集-缓存队列-实时处理计算-数据展现。本文就用阿里云产品简单实现了一个实时处理的方案。

一,总体架构
1

按照数据流向
数据采集:flume(配置故障转移)
缓存队列:datahub
https://help.aliyun.com/product/53345.html?spm=5176.7618386.3.4.cigK2v
数据计算:阿里流计算(StreamCompute)
https://help.aliyun.com/video_list/54212.html?spm=5176.7618386.3.2.COgP6l
数据落地:rds(mysql)
https://help.aliyun.com/document_detail/26092.html?spm=5176.7841871.6.539.9FTjxU
数据展现:Quick-BI
https://data.aliyun.com/product/bi?spm=5176.8142029.388261.284.spvIS0
或者大屏显示 DATA-V
https://data.aliyun.com/visual/datav?spm=5176.8142029.388261.283.spvIS0

二,搭建过程

1,flume配置搭建
flume在数据采集的开源框架中还是比较常用的,但是在采集输送到datahub中有可能网络断了或者服务器挂了。那这里配置了故障转移,如图,其中sink1和sink2为上面架构中的agentA和agentB.把agentA和agentB分别部署在两台服务器上。
2

在搭建flume时需要安装DatahubSink插件,参考https://help.aliyun.com/knowledge_detail/42843.html
那看下配置文件


# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1

# Describe/configure the source这里监控一个文件变化,写了一个定时脚本每秒插入一条
a1.sources.r1.type = exec
a1.sources.r1.channels=c1
a1.sources.r1.command=tail -F /usr/local/shangdan/test.txt

#define sinkgroups,在这里配置故障转移的sink组
a1.sinkgroups=g1
a1.sinkgroups.g1.sinks=k1 k2
a1.sinkgroups.g1.processor.type=failover
a1.sinkgroups.g1.processor.priority.k1=10//这里设置sink的优先级,优先发送到级别高的sink里
a1.sinkgroups.g1.processor.priority.k2=5
a1.sinkgroups.g1.processor.maxpenalty=10000

#define the sink 1,发送到agentA
a1.sinks.k1.type=avro
a1.sinks.k1.hostname=agentA的ip
a1.sinks.k1.port=5555

#define the sink 2 ,发送到agentB
a1.sinks.k2.type=avro
a1.sinks.k2.hostname=agentB的ip
a1.sinks.k2.port=5555


# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel=c1
~

agentA和agentB的配置文件除了IP地址不一样,其他完全一致,这里贴其中一个

A single-node Flume configuration for Datahub
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels=c1
a1.sources.r1.bind= agentA的ip
a1.sources.r1.port= 5555
# Describe the sink
a1.sinks.k1.type = com.aliyun.datahub.flume.sink.DatahubSink
a1.sinks.k1.datahub.accessID = ******
a1.sinks.k1.datahub.accessKey = **********
a1.sinks.k1.datahub.endPoint = http://dh-cn-hangzhou.aliyun-inc.com
a1.sinks.k1.datahub.project = shangdantest
a1.sinks.k1.datahub.topic = databubtest
a1.sinks.k1.serializer = DELIMITED
a1.sinks.k1.serializer.delimiter = ,//这里配置数据的分隔符
a1.sinks.k1.serializer.fieldnames = line//配置数据的字段
a1.sinks.k1.batchSize = 1
a1.sinks.k1.serializer.charset = UTF-8
a1.sinks.k1.shard.number = 1
a1.sinks.k1.shard.maxTimeOut = 60
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 1000
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

三台服务配置完成后启动flume(先启动agentA和agentB)预期结果是agent1发送数据到agentA(优先级高的),如果停止agentA服务,会自动转换发送到agentB。重启agegtA的服务后,再次切回到agentA。
如图:正常启动数据正常传输经过agent1-agentB-datahub
3

此时,停掉agentA服务,日志报错,故障转移。
4

重启agentA服务,恢复到之前状态,切回到sink1
5

2,datahub创建,
在datahub控制台创建项目和topic,
设置分片和生命周期,具体方法见链接
https://help.aliyun.com/document_detail/47448.html?spm=5176.doc47443.6.584.UrSX1A
datahub中看到有flume传过来的数据
6

3,配置阿里流计算

登录阿里流计算控制台

注册数据源datahub/rds(也支持阿里其他类型数据源)-编写流计算脚本-调试-上线-启动

如图先注册数据源供脚本使用。必须要有数据来源表和数据结果表。
8

在编写脚本时,可以直接引用表,会自动插入表结构和配置信息,非常方便
9

那开始编写脚本必须包括三部分
1,创建数据来源表,这里是datahub表
2,创建数据结果表,这里是rds表
3,将来源表数据写入结果表,并进行计算

如图
10
三、测试

   脚本编写完毕,点击上方【调试】,可以自己先准备一些数据上传测试。也可以直接线上测试,点击上面【上线】,上线成功后在【运维】中能看到项目,点击启动,项目启动几秒就工作了如图:

11

 然后可以看到监控状态,计算延迟,数据是否倾斜等指标,也有更详细的链路可以查看

12

最后,我们把整个流程全部启动,到rds中看结果如图
13

当然,如果希望源源不断的流数据保存下来称为静态的数据,作为后续业务分析统计等用途,在datahub控制台可以直接配置归档到大数据计算服务(Maxcompute)中,直接入库为表数据。
如图
14
需要在Maxcompute中创建好对应表即可自动归档存储。详细配置如下
https://help.aliyun.com/document_detail/47453.html?spm=5176.doc47439.6.555.3GNrRs

好神奇,几句sql数据就源源不断的流过来,那么前端或者其他业务层可以过来拿数据展示了,数据还可以界面化配置归档入库,十分方便。如果有复杂逻辑计算的,可以申请开通流计算的udf功能,这样看来,学好sql和java,走遍天下都不怕。
数据可视化部分可以参考使用阿里云产品dataV,实现类似双十一大屏效果,也可以使用产品Quick-BI做实时报表。

有对大数据技术感兴趣的,可以加笔者的微信 wx4085116.目前笔者已经从阿里离职,博客不代表阿里立场。笔者开了一个大数据培训班。有兴趣的加我。

相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
目录
相关文章
|
8月前
|
分布式计算 DataWorks 大数据
阿里云ODPS的个人收获思考
在接触阿里云ODPS过程中,我深入了解了MaxCompute和DataWorks等产品。MaxCompute强大的数据处理能力显著提升了我的工作效率,而DataWorks的一站式开发与治理平台简化了数据流程管理。通过实践,我不仅掌握了高效的SQL编写与数据挖掘技巧,还提升了团队协作意识与大数据思维,为未来挑战打下了坚实基础。
174 3
|
8月前
|
存储 分布式计算 大数据
【赵渝强老师】阿里云大数据存储计算服务:MaxCompute
阿里云MaxCompute是快速、全托管的TB/PB级数据仓库解决方案,提供海量数据存储与计算服务。支持多种计算模型,适用于大规模离线数据分析,具备高安全性、低成本、易用性强等特点,助力企业高效处理大数据。
397 0
|
8月前
|
数据采集 人工智能 大数据
10倍处理效率提升!阿里云大数据AI平台发布智能驾驶数据预处理解决方案
阿里云大数据AI平台推出智能驾驶数据预处理解决方案,助力车企构建高效稳定的数据处理流程。相比自建方案,数据包处理效率提升10倍以上,推理任务提速超1倍,产能翻番,显著提高自动驾驶模型产出效率。该方案已服务80%以上中国车企,支持多模态数据处理与百万级任务调度,全面赋能智驾技术落地。
1150 0
|
11月前
|
数据采集 机器学习/深度学习 人工智能
面向 MoE 和推理模型时代:阿里云大数据 AI 产品升级发布
2025 AI 势能大会上,阿里云大数据 AI 平台持续创新,贴合 MoE 架构、Reasoning Model 、 Agentic RAG、MCP 等新趋势,带来计算范式变革。多款大数据及 AI 产品重磅升级,助力企业客户高效地构建 AI 模型并落地 AI 应用。
|
6月前
|
人工智能 分布式计算 DataWorks
阿里云大数据AI产品月刊-2025年8月
阿里云大数据& AI 产品技术月刊【2025年 8 月】,涵盖 8 月技术速递、产品和功能发布、市场和客户应用实践等内容,帮助您快速了解阿里云大数据& AI 方面最新动态。
484 2
|
6月前
|
存储 分布式计算 资源调度
【赵渝强老师】阿里云大数据MaxCompute的体系架构
阿里云MaxCompute是快速、全托管的EB级数据仓库解决方案,适用于离线计算场景。它由计算与存储层、逻辑层、接入层和客户端四部分组成,支持多种计算任务的统一调度与管理。
511 1
|
8月前
|
人工智能 分布式计算 DataWorks
多模态数据处理新趋势:阿里云ODPS技术栈深度解析与未来展望
阿里云ODPS技术栈通过MaxCompute、Object Table与MaxFrame等核心组件,实现了多模态数据的高效处理与智能分析。该架构支持结构化与非结构化数据的统一管理,并深度融合AI能力,显著降低了分布式计算门槛,推动企业数字化转型。未来,其在智慧城市、数字医疗、智能制造等领域具有广泛应用前景。
674 6
多模态数据处理新趋势:阿里云ODPS技术栈深度解析与未来展望