Tablestore+Delta Lake(快速开始)

2021-12-18 02:37:17

作者：王卓然花名琸然阿里云存储服务技术专家

背景介绍

近些年来HTAP(Hybrid transaction/analytical processing)的热度越来越高，通过将存储和计算组合起来，既能支持传统的海量结构化数据分析，又能支持快速的事务更新写入，是设计数据密集型系统的一个成熟的架构。
表格存储(Tablestore)是阿里云自研的 NoSQL 多模型数据库，提供海量结构化数据存储以及快速的查询和分析服务(PB 级存储、千万 TPS 以及毫秒级延迟)，借助于表格存储的底层引擎，能够很好的完成OLTP场景下的需求。Delta Lake类似于支持Delta的Data Lake(数据湖)，使用列存来存base数据，行的格式存储新增delta数据，进而做到支持数据操作的ACID和CRUD，完全兼容Spark的大数据生态，通过结合Delta Lake和Spark生态，能够很好的完成OLAP场景下的需求。下图展示的是Tablestore和Delta Lake结合的HATP场景的一个简要的逻辑结构图，有关结构化大数据分析平台设计的更多细节和干货，可以参阅文章结构化大数据分析平台设计。

准备工作

登录阿里云E-MapReduce控制台
创建Hadoop集群(若已创建，请跳过)
确保将Tablestore实例部署在E-MapReduce集群
相同的VPC环境下

步骤一创建Tablestore源表

详细开通步骤请参考官方文档，本文demo中所创建出来的表名为Source，表的Schema如下图所示，该表有PKString和PkInt两个主键，类型分别为String和Interger。

为表Source建立一个增量通道，如下图所示，通道列表里面会显示该通道的名字、ID以及类型。

技术注解:

通道服务（Tunnel Service）是基于Tablestore数据接口之上的全增量一体化服务，包含三种通道类型：

全量：对数据表中历史存量数据消费处理

增量：对数据表中新增数据消费处理
全量加增量：先对数据表总历史存量数据消费，之后对新增数据消费
通道服务的详细介绍可查询Tablestore官网文档。

步骤二获取相关jar包并上传到hadoop集群

获取环境依赖的JAR包。

Jar包 | 获取方法 |
------- | ------- |
emr-tablestore-X.X.X.jarX.X.X: Since 1.9.0+ |Maven 库中下载：https://mvnrepository.com/artifact/com.aliyun.emr/emr-tablestore
tablestore-X.X.X-jar-with-dependencies.jar |下载 EMR SDK 相关的Tablestore依赖包。https://repo1.maven.org/maven2/com/aliyun/openservices/tablestore/5.3.0/tablestore-5.3.0-jar-with-dependencies.jar

在集群管理页面，单击已创建的Hadoop集群的集群ID ，进入集群与服务管理页面。
在左侧导航树中选择主机列表，然后在右侧查看Hadoop集群中emr-header-1主机的IP信息。
在SSH客户端中新建一个命令窗口，登录Hadoop集群的emr-header-1主机。
上传所有JAR包到emr-header-1节点的某个目录下。

步骤三运行Spark Streaming作业

以一个基于emr demo修改的代码为样例，编译生成JAR包，JAR包需要上传到Hadoop集群的emr-header-1主机中(参见步骤二)，完整的代码由于改动较大，不在本文中一一说明，后续会合到emr demo官方项目中。
该样例以Tablestore表作为数据源，通过结合Tablestore CDC技术，Tablestore Streaming Source和Delta Sink，演示的是TableStore到Delta Lake的一个完整链路。
按以下命令，启动spark streaming作业，开启一个实时同步Tablestore Source表中数据到Delta Lake Table的监听程序。

各个参数说明如下：

参数	参数说明
com.aliyun.emr.example.spark.sql.streaming.DeltaTableStoreCDC	所要运行的主程序类
emr-tablestore-X.X.X-SNAPSHOT.jar	包含Tablestore source的jar包
tablestore-X.X.X-jar-with-dependencies.jar	EMR SDK 相关的Tablestore依赖包
examples-X.X.X-shaded.jar	基于EMR demo修改的包(包含主程序类)
instance	Tablestore实例名
tableName	Tablestore表名
tunnelId	Tablestore表的通道Id
accessKeyId	Tablestore的accessKeyId
accessKeySecret	Tablestore的秘钥
endPoint	Tablestore实例的endPoint
maxOffsetsPerChannel	Tablestore通道 Channel在每个Spark Batch周期内同步的最大数据条数，默认10000。
catalog	同步的列名，详见Catalog字段说明

步骤四数据CRUD示例

首先在TableStore里插入两行，本次示例中，我们建了8列的同步列，包括两个主键(PkString, PkInt)和六个属性列(col1, col2, col3, timestamp, col5和col6)。由于表格存储是Free-Schema的结构，我们可以任意的插入属性列，TableStore的Spark Source会自动的做属性列的筛选。如下面两张图所示，在插入两行数据后，Delta Table中同步也可以马上读取到两行，且数据一致。