5分钟迅速搭建云上Lambda大数据分析架构-阿里云开发者社区

开发者社区> 阿里云EMR> 正文
登录阅读全文

5分钟迅速搭建云上Lambda大数据分析架构

简介: 主要介绍基于 Tablestore 的数据变更实时捕获订阅能力,实现云上Lambda 架构的轻量化实现数据的实时和离线处理。演示模拟了一个电商订单场景,通过流计算实现订单大屏的场景,做到海量订单实时注入的同时,进行10s的订单统计聚合以及交易金额统计并做实时的大屏幕展示

作者:朱晓然,Tablestore存储服务技术专家


背景
Spark 中国社区联合阿里云 EMR 技术交流群,Tablestore 技术交流群举办了一场联合技术直播。直播的话题是“海量结构化数据的实时计算和处理”,主要介绍基于 Tablestore 的数据变更实时捕获订阅能力,实现云上Lambda 架构的轻量化实现。在直播中有一个demo环节,本篇文章会提供demo环节的简单操作步骤,方便大家后续在阿里云上搭建和demo场景类似的一整套架构,实现数据的实时和离线处理。

演示场景介绍
演示模拟了一个电商订单场景,通过流计算实现订单大屏的场景,做到海量订单实时注入的同时,进行10s的订单统计聚合以及交易金额统计并做实时的大屏幕展示。整个订单的大屏幕样例如下:
image.png

大屏我们使用阿里云的 DATAV 对接 Tablestore数据源来实现,那么下面我们就具体看看从订单的原始数据到结果大屏数据的产生过程以及操作步骤。

整套后台的架构大体如下:
image.png

  1. 在ecs,或者本地模拟一个订单生成器,实时的注入订单数据到 Tablestore 中。
  2. 在 Tablestore 控制台创建通道
  3. 在 EMR 控制台购买 Spark 集群
  4. 下载最新的 EMR SDK
  5. 执行下面提供的建表语句和SQL命令实现实时计算,结果表会写回 Tablestore中。
  6. 通过 DATAV 进行实时大屏展示结果表数据

操作步骤一:登陆阿里云官网 Tablestore 控制台进行实例和表创建
image.png

创建实例后,可以创建一张表,表主键schema如下:
image.png

启动客户端注入程序随机写入数据,样例数据如下:
image.png

Tablestore 产品是 Serverless的形态,用户使用无需购买大小或者规格,产品回根据业务做自动水平扩展。

操作步骤二:登陆阿里云官网 EMR 控制台购买Spark集群
Spark的集群规模可以根据业务需求灵活选取,我们实测三节点,可以轻松的实时消费100w/s的数据做聚合计算哟!
image.png

操作步骤三:登陆EMR集群执行作业脚本
登陆EMR的master节点,执行下面命令启动流任务:

1.启动stream sql交互
在EMR 官网获取最新版本EMR sdk(1.8)
streaming-sql --driver-class-path emr-datasources_shaded_2.11-1.8.0.jar --jars emr-datasources_shaded_2.11-1.8.0.jar --master yarn-client --num-executors 8 --executor-memory 2g --executor-cores 2

2.创建streaming source 表
DROP TABLE IF EXISTS ots_order_test;
CREATE TABLE ots_order_test
USING tablestore
OPTIONS(
endpoint="填写Tablestore VPC的地址",
access.key.id="",
access.key.secret="",
instance.name="",
table.name="",
tunnel.id="在Tablestore控制台查找对应想消费通道ID",
catalog='{"columns": {"UserId": {"col": "UserId", "type": "string"}, "OrderId": {"col": "OrderId", "type": "string"},"price": {"cols": "price", "type": "long"}, "timestamp": {"cols": "timestamp", "type": "long"}}}'
);

3.创建streaming sink表
DROP TABLE IF EXISTS ots_order_sink_test;
CREATE TABLE ots_order_sink_test
USING tablestore
OPTIONS(
endpoint="",
access.key.id="",
access.key.secret="",
instance.name="",
table.name="",
tunnel.id="",
catalog='{"columns": {"begin": {"col": "begin", "type": "string"},"end": {"col": "end", "type": "string"}, "count": {"col": "count", "type": "long"}, "totalPrice": {"col": "totalPrice", "type": "long"}}}'
);

4.创建Streaming作业
CREATE SCAN ots_table_stream on ots_order_test USING STREAM OPTIONS ("maxoffsetsperchannel"="10000");
CREATE STREAM job1
options(
checkpointLocation='/tmp/spark/cp/test1',
outputMode='update'
)
insert into ots_order_sink_test
SELECT CAST(window.start AS String) AS begin, CAST(window.end AS String) AS end, count(*) AS count, sum(price) AS totalPrice FROM ots_table_stream GROUP BY window(to_timestamp(timestamp / 1000000000), "10 seconds");

最后实验有任何问题,或者希望做技术交流的同学欢迎加入我们的技术交流群(钉钉:23307953 或者11789671),来与我们一起探讨。
原文链接:https://yq.aliyun.com/articles/721294

image.png

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,技术专家直播,问答区数个Spark技术同学每日在线答疑,只为营造纯粹的Spark氛围,欢迎钉钉扫码加入!
二维码.JPG

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

分享:
+ 订阅

阿里巴巴开源大数据技术团队成立阿里云EMR技术圈, 每周推送前沿技术文章,直播分享经典案例、在线答疑,营造纯粹的开源大数据氛围,欢迎加入!加入钉钉群聊阿里云E-MapReduce交流2群,点击进入查看详情 https://qr.dingtalk.com/action/joingroup?code=v1,k1,cNBcqHn4TvG0iHpN3cSc1B86D1831SGMdvGu7PW+sm4=&_dt_no_comment=1&origin=11

官方博客
官网链接