背景
Spark 中国社区联合阿里云 EMR 技术交流群，Tablestore 技术交流群举办了一场联合技术直播。直播的话题是“海量结构化数据的实时计算和处理”，主要介绍基于 Tablestore 的数据变更实时捕获订阅能力，实现云上Lambda 架构的轻量化实现。在直播中有一个demo环节，本篇文章会提供demo环节的简单操作步骤，方便大家后续在阿里云上搭建和demo场景类似的一整套架构，实现数据的实时和离线处理。
演示场景介绍
演示模拟了一个电商订单场景，通过流计算实现订单大屏的场景，做到海量订单实时注入的同时，进行10s的订单统计聚合以及交易金额统计并做实时的大屏幕展示。整个订单的大屏幕样例如下：
大屏我们使用阿里云的 DATAV 对接 Tablestore数据源来实现，那么下面我们就具体看看从订单的原始数据到结果大屏数据的产生过程以及操作步骤。
整套后台的架构大体如下：
- 在ecs，或者本地模拟一个订单生成器，实时的注入订单数据到 Tablestore 中。
- 在 Tablestore 控制台创建通道
- 在 EMR 控制台购买 Spark 集群
- 下载最新的 EMR SDK
- 执行下面提供的建表语句和SQL命令实现实时计算，结果表会写回 Tablestore中。
- 通过 DATAV 进行实时大屏展示结果表数据
操作步骤一：登陆阿里云官网 Tablestore 控制台进行实例和表创建
创建实例后，可以创建一张表，表主键schema如下:
启动客户端注入程序随机写入数据，样例数据如下：
Tablestore 产品是 Serverless的形态，用户使用无需购买大小或者规格，产品回根据业务做自动水平扩展。
操作步骤二：登陆阿里云官网 EMR 控制台购买Spark集群
Spark的集群规模可以根据业务需求灵活选取，我们实测三节点，可以轻松的实时消费100w/s的数据做聚合计算哟！
操作步骤三：登陆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），来与我们一起探讨。