前言
从EMR-3.21.0 版本开始将提供Spark Streaming SQL的预览版功能，支持使用SQL来开发流式分析作业。结果数据可以实时写入Tablestore。
本文以LogHub为数据源，收集ECS上的日志数据，通过Spark Streaming SQL进行聚合后，将流计算结果数据实时写入Tablestore，展示一个简单的日志监控场景。
场景设计
假设有一个商品表Goods，商品信息开放给用户浏览，用户浏览完以后会产生以下格式的日志数据：
{
"RequestId":"c85df119-f6db-449f-89bb-6773d2468f89",
"Time":2019-07-30 12:05:28,
"GoodsName":"shoes",
"OperationType":"query"
}
AI 代码解读
我们需要将原始日志数据，根据GoodsName、OperationType和时间，聚合成一分钟一个点的监控数据。用于监控各商品的访问情况。
最终需要的监控数据格式如下：
| GoodsName | OperationType | Time | OperationCount |
|---|---|---|---|
| shoes | query | 2019-07-30 12:05 | 5060 |
| shoes | query | 2019-07-30 12:06 | 8001 |
| shoes | query | 2019-07-30 12:07 | 9607 |
过程如下图所示，比较简单。
技术选型
本文主要来看一下结果数据存放数据库的选型。对于本文的监控场景，结果数据的量级，取决于商品数量。真实情况下，可能还要增加商品的规格、颜色等监控指标，结果数据量会比较大。此外，对于时间粒度比较小的监控数据，一般都只需要保留最近的，时间比较久的历史数据需要删除。
将Tablestore与传统关系型数据库MySQL进行对比，Tablestore有以下优势：
- 支持海量数据，无缝扩展
表格存储通过数据分片和负载均衡技术，实现了无缝扩展。 - 支持数据自动过期
数据生命周期（Time To Live，简称 TTL）是数据表的一个属性，即数据的存活时间，单位为秒。表格存储会在后台对超过存活时间的数据进行清理，以减少用户的数据存储空间，降低存储成本。监控场景下比较适用，不需要手动去删除数据。
预备工作
创建EMR集群
开通EMR之前，先要对云账号进行实名认证，然后创建默认的EMR角色并授予 AliyunEMRDefaultRole和AliyunEmrEcsDefaultRole这两个角色。
EMR暂时还不支持在官网控制台上配置写入Tablestore的任务，需要登录到MER集群的机器上去操作，所以开通EMR集群的时候，请选择自定义购买，并在最后一步打开挂载公网、远程登录、密码方式登录，自己设置一个密码，如下图。 
开通完以后，进入到ECS控制台，会看到有一台Master节点机器有弹性IP。后面步骤中，操作EMR Spark Streaming需要使用这个IP地址远程登录到机器上。
开通ECS和日志服务
详细步骤请参考官方文档。
其中ilogtail配置如下
日志收集到LogHub以后，可以通过官网控制台查看。本文示例中收集到的数据如下：
创建Tablestore结果表
详细开通步骤请参考官方文档。最终创建出来的表Schema如下：
Count列，作为属性列，不需要定义在主键中。
数据处理流程
1、下载支持数据源需要的jar包
2、进入streaming-sql执行环境
下载完jar包以下，在包所在目录执行以下命令进入交互式开发环境。
streaming-sql --master yarn-client --jars ./emr-datasources_shaded_2.11-1.7.0.jar --driver-class-path ./emr-datasources_shaded_2.11-1.7.0.jar
AI 代码解读
注意：jar包要上传到集群机器上，不支持远程引用oss文件。
3、创建LogHub数据源表
创建LogHub数据源表之前，需要手动开通日志服务的project和logStore，并将日志数据收集好，具体参考【预备工作】一节中，开通LogHub的官方文档。
然后再到EMR集群机器的交互式执行环境中创建LogHub数据源表，示例如下：
CREATE DATABASE IF NOT EXISTS helloemr;
USE helloemr;
CREATE TABLE loghub_source(GoodsName string,OperationType string,RequestId string,__time__ timestamp)
USING loghub
OPTIONS (
sls.project = '{your project name}',
sls.store = '{store name}',
access.key.id = '{access-key}',
access.key.secret = '{access-key-secret}',
endpoint = 'http://{your project name}.cn-hangzhou-intranet.log.aliyuncs.com');
AI 代码解读
上面的sql语句，创建一个database为helloemr，tablename为loghub_source的表，shema中有GoodsName、OperationType、Time、RequestId、__time__五个字段，其中__time__字段是在配置ilogtail的时候解析日志中的Time字段得到的，等同于Time字段。
需要注意的是，Endpoint请使用内网域名，公网域名速度上会慢很多。
4、创建Tablestore结果表
同样的，创建Tablestore结果表之前，也要先到官网控制台建到tablestore的实例和表。
创建EMR的Tablestore结果表，示例如下
CREATE DATABASE IF NOT EXISTS helloemr;
USE helloemr;
CREATE TABLE tablestore_sink
USING tablestore
OPTIONS(
endpoint="https://sparkStreaming.cn-hangzhou.vpc.tablestore.aliyuncs.com",
access.key.id="{access-key}",
access.key.secret="{access-key-secret}",
table.name="tablestore_sink",
instance.name="sparkStreaming",
catalog='{"columns":{"GoodsName":{"col":"GoodsName","type":"string"},"OperationType":{"col":"OperationType","type":"string"},"Time":{"col":"Time","type":"long"},"Count":{"col":"Count","type":"long"}}}');
AI 代码解读
以上sql代码，创建了一个表名为tablestore_sink的表，实例名是sparkStreaming。注意：Endpoint请使用vpc域名。
创建成功以后，使用desc tablestore_sink; 查看表结构如下。
5、结果数据写入到Tablestore
通过GoodsName、OperationType聚合一分钟内的请求次数。这里要用到Spark Streaming的滚动窗口函数，取window.start作为聚合后的时间。
USE helloemr;
SET streaming.query.name=loghub_source;
SET spark.sql.streaming.checkpointLocation.loghub_source=/home/helloemr;
SET spark.sql.streaming.query.trigger.loghub_source=ProcessingTime;
SET spark.sql.streaming.query.trigger.intervalMs.loghub_source=10000;
INSERT INTO tablestore_sink
SELECT GoodsName,OperationType,count(*) as Count,to_unix_timestamp(window.start, 'yyyy-MM-dd HH:mm:ss') as Time from loghub_source
where delay(__time__)<"2 minute"
GROUP BY TUMBLING (__time__, interval 1 minute),GoodsName,OperationType;
AI 代码解读
示例SQL中只填了几个必填参数，具体可以参考作业模板。其中checkpointLocation代表本次流式查询作业的checkpoint路径，需要设置一个绝对路径值。
最终，Tablestore中的结果数据如下：
性能调优
Spark Streaming 是基于Spark的流式处理引擎，其基本原理是把输入数据以某一时间间隔批量的处理，当批处理间隔缩短到秒级时，便可以用于处理实时数据流。它本质上是微批处理。
使用上一节的示例代码，实际测试下来，一次作业需要耗费1秒左右的时间。在实际实例中，由于源表、目标数据表、数据量大小以及Sql的复杂程度不同，耗费的时间也会不同。
数据读取
本文使用LogHub为数据源，Shard数量多，Spark Streaming的作业并发度也会多，但需要设置合理的Shard数，具体请参考日志服务分区设置。
同时，创建LogHub数据源表的时候，请使用内网Endpoint。
Spark Streaming作业调优
设置合理的批处理时间
trigger.intervalMs代表批次间隔，单位毫秒，默认为0L。运行任务的时候，当间隔时间比一次任务的运行时间短的时候，任务会打印WARN日志。一般这个值的大小如果能够使得Streaming作业刚好处理好上一个的批处理的数据，那么这个就是最优值。
增加作业资源
EMR官网控制台里面有的监控大盘，可以看到作业占用的资源情况，可以根据实际情况调整分配给作业的资源大小。
数据写入
对于表格存储(Tablestore)来说，合理的主键设计，是提高写入性能的关键因素。具体可以参考表格存储最佳实践。一个设计良好的主键，需要避免访问压力集中在一个小范围的连续的分片键上，也就是说避免热点分片。设计良好的表结构，整张表的访问压力能够均匀的分散在各个分片上，这样才能充分利用后端服务器的能力。
如果是新建的Tablestore表，而且数据写入量比较大，最好联系下@表格存储技术支持，对表进行预分区。可以提高初始状态下的写入性能。
欢迎加入
表格存储(Tablestore)推出了很多贴近用户场景的文章与示例代码，欢迎大家加入我们的钉钉公开交流群一起讨论，群号：11789671。
