前言

从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"
}

我们需要将原始日志数据，根据GoodsName、OperationType和时间，聚合成一分钟一个点的监控数据。用于监控各商品的访问情况。
最终需要的监控数据格式如下：

过程如下图所示，比较简单。

技术选型

本文主要来看一下结果数据存放数据库的选型。对于本文的监控场景，结果数据的量级，取决于商品数量。真实情况下，可能还要增加商品的规格、颜色等监控指标，结果数据量会比较大。此外，对于时间粒度比较小的监控数据，一般都只需要保留最近的，时间比较久的历史数据需要删除。
将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包

下载地址
https://github.com/aliyun/aliyun-emapreduce-sdk/blob/master-2.x/jars/datasources/latest/emr-datasources_shaded_2.11-1.7.0.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

注意：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');

上面的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"}}}');

以上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;

示例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。