使用EMR DataFrame 批处理 Tablestore

2022-05-09 03:00:41

使用Spark的DataFrame方式访问表格存储，并在本地和集群上分别进行运行调试。

前提条件

了解Spark访问表格存储的依赖包，并在使用时通过maven方式引入项目中。
- Spark相关：spark-core、spark-sql、spark-hive
- Spark Tablestore connector：emr-tablestore-.jar
- Tablestore Java SDK：tablestore--jar-with-dependencies.jar
其中表示相应依赖包的版本号，请以实际为准。
已在表格存储侧创建数据表，详情请参见概述。

快速开始

通过项目样例了解快速使用批计算的操作。

从GitHub下载项目样例的源码，具体下载路径请参见TableStoreSparkDemo。项目中包含完整的依赖和使用样例，具体的依赖请参见项目中的pom文件。
阅读TableStoreSparkDemo项目的README文档，并安装最新版的Spark Tablestore connector和Tablestore Java SDK到本地maven库。

说明：Spark Tablestore connector正式版发布以月为周期，目前最新版尚未正式发布，请先使用项目附带的预览版，正式发布后，本文也会进行更新，敬请期待。预览版和正式版只是版本号的区别，相互兼容，业务代码逻辑无需改动。
修改Sample代码
以TableStoreBatchSample为例，对此示例代码的核心代码说明如下：
- format("tablestore")表示使用ServiceLoader方式加载Spark Tablestore connector，具体配置请参见项目中的META-INF.services。
- instanceName、tableName、endpoint、accessKeyId、accessKeySecret分别表示表格存储的实例名称、数据表名称、实例endpoint、阿里云账号的AccessKey ID和AccessKey Secret。
- catalog是一个json串，包含字段名和类型，如下示例中的数据表有salt（Long类型）、UserId（String类型）、OrderId（String类型）、price（Double类型）和timestamp（Long类型）五个字段。最新版本中支持使用Schema方式替换catalog的配置，请根据实际选择。
- split.size.mbs表示每个Split的切分大小，默认值为100，单位为MB，可不配置。此值越小产生的Split会越多，对应Spark的Task也会越多。
```
val df = sparkSession.read
.format("tablestore")
.option("instance.name", instanceName)
.option("table.name", tableName)
.option("endpoint", endpoint)
.option("access.key.id", accessKeyId)
.option("access.key.secret", accessKeySecret)
.option("split.size.mbs", 100)
.option("catalog", dataCatalog)
// 最新版本支持使用Schema方式替换catalog的配置。
//.schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG")
.load()

val dataCatalog: String =
  s"""
     |{"columns": {
     |    "salt": {"type":"long"},
     |    "UserId": {"type":"string"},
     |    "OrderId": {"type":"string"},
     |    "price": {"type":"double"},
     |    "timestamp": {"type":"long"}
     | }
     |}""".stripMargin
```

运行调试

根据需求修改示例代码后，可在本地或者通过Spark集群进行运行调试。以TableStoreBatchSample为例说明调试过程。

本地调试

以Intellij IDEA为例说明。

说明：本文测试使用的环境为Spark 2.4.3、Scala 2.11.7和Java SE Development Kit 8，如果使用中遇到问题，请联系表格存储技术支持。

在系统参数中，配置实例名称、数据表名称、实例endpoint、阿里云账号的AccessKey ID和AccessKey Secret等参数。您也可以自定义参数的加载方式。
选择include dependencies with "provided" scope，单击OK。
运行示例代码程序。

通过Spark集群调试

以spark-submit方式为例说明。示例代码中的master默认为local[*]，在Spark集群上运行时可以去掉，使用spark-submit参数传入。

执行mvn -U clean package命令打包，包的路径为target/tablestore-spark-demo-1.0-SNAPSHOT-jar-with-dependencies.jar。

上传包到Spark集群的Driver节点，并使用spark-submit提交任务。

spark-submit --class com.aliyun.tablestore.spark.demo.batch.TableStoreBatchSample --master yarn tablestore-spark-demo-1.0-SNAPSHOT-jar-with-dependencies.jar <ots-instanceName> <ots-tableName> <access-key-id> <access-key-secret> <ots-endpoint>