访问Tablestore数据源

本文介绍如何使用AnalyticDB MySQL湖仓版（3.0）Spark通过访问 Tablestore 数据。

前提条件

已创建AnalyticDB MySQL湖仓版（3.0）集群。具体操作，请参见创建湖仓版（3.0）集群。
已创建数据库账号。
- 如果您是通过阿里云账号访问，只需创建高权限账号。具体操作，请参见创建高权限账号。
- 如果是通过RAM用户访问，需要创建高权限账号和普通账号并且将RAM用户绑定到普通账号上。具体操作，请参见创建数据库账号和绑定RAM用户与数据库账号。
已创建Job型资源组。具体操作，请参见新建资源组。
已创建与AnalyticDB MySQL湖仓版（3.0）集群同地域的 TableStore 实例。具体操作，请参见创建 Tablestore 实例。

操作步骤

下载AnalyticDB MySQL Spark访问 Tablestore 依赖的Jar包。下载链接 [待赵培杰上传 icms 后添加]
在pom.xml文件的dependencies中添加依赖项

<dependency>
  <groupId>com.aliyun.emr</groupId>
  <artifactId>emr-tablestore</artifactId>
  <version>2.2.0</version>
  <scope>provided</scope>
</dependency>

编写如下示例程序来访问Redis，并进行编译打包。本文生成的Jar包名称为 spark-tablestore.jar。示例代码如下：

package com.aliyun.spark

import com.aliyun.openservices.tablestore.hadoop.TableStore
import org.apache.spark.sql.SparkSession
import scala.collection.mutable

object SparkOts {
  def main(args: Array[String]): Unit = {
    // 您访问的talbestore endpoint
    val endpoint = args(0)
    // 您想要连接的表名
    val tableName = args(1)
    // 您的tablestore实例名称
    val instanceName = args(2)
    // 您的ACCESS KEY
    val accessKeyId = sys.env.get("ACCESS_KEY_ID")
    // 您的ACCESS KEY SECRET
    val accessKeySecret = sys.env.get("ACCESS_KEY_SECRET")
    
    // 您访问的tablestore表数据目录
    val dataCatalog = 
      """
        |{"columns":{
        |    "PkString": {"type":"string"},
        |    "PkInt": {"type":"long"},
        |    "col1": {"type":"string"},
        |    "col2": {"type":"long"},
        |    "col3": {"type":"binary"},
        |    "timestamp": {"type":"long"},
        |    "col5": {"type":"double"},
        |    "col6": {"type":"boolean"}
        |  }
        |}
      """.stripMargin
    
    val options = new mutable.HashMap[String, String]()
    options.put("endpoint", endpoint)
    options.put("access.key.id", accessKeyId)
    options.put("access.key.secret", accessKeySecret)
    options.put("table.name", tableName)
    options.put("instance.name", instanceName)
    options.put("catalog", dataCatalog)

    val spark = SparkSession.builder().getOrCreate()
    // 读取
    val df = spark.read.format("tablestore").options(options).load()
    
    df.show()
    // 写入
    df.write.format("tablestore").options(options).save()

    TableStore.shutdown()
    spark.stop()
  }
}

将步骤1下载的Jar包和示例程序spark-tablestore.jar上传至OSS。具体操作，请参见简单上传。
进入Spark Jar开发。
1. 登录云原生数据仓库AnalyticDB MySQL控制台。
2. 在页面左上角，选择集群所在地域。
3. 在左侧导航栏，单击集群列表。
4. 在湖仓版（3.0）页签下，单击目标集群ID。
5. 在左侧导航栏，单击作业开发> Spark Jar 开发。
在编辑器窗口上方，选择Job型资源组和作业类型。本文以Batch类型为例。
在编辑器中输入以下作业内容。

{
    "args": [
        -- Tablestore实例的VPC连接地址。在目标实例的实例管理页面的实例详情区域，可查看到各连接类型的地址和端口号。
        "https://x00vc21i***.<region_id>.vpc.tablestore.aliyuncs.com",
        -- 访问的Tablestore表名
	    "spark_****",
        -- 访问的Tablestore实例名
	    "x00vc21i***"
    ],
    "file": "oss://<bucket_name>/spark-tablestore.jar",
	"jars": [
		"oss://<bucket_name>/emr-tablestore-2.3.0-SNAPSHOT.jar",
	    "oss://<bucket_name>/tablestore-5.10.3-jar-with-dependencies.jar"
	],
    "name": "spark-on-ots",
    "className": "com.aliyun.spark.SparkOnTablestore",
    "conf": {
        "spark.driver.resourceSpec": "medium",
        "spark.executor.instances": 2,
        "spark.executor.resourceSpec": "medium",
        "spark.kubernetes.driverEnv.ACCESS_KEY_ID": "LTAI5tLBmYLvf*******",
        "spark.kubernetes.driverEnv.ACCESS_KEY_SECRET": "A4KuIceJjlARWKzA6nVbbi*******",
	      "spark.hadoopRDD.ignoreEmptySplits": false
    }
}

参数说明如下。

参数	说明
file	Spark作业主文件的存储位置。主文件是入口类所在的Jar包或者Python的入口执行文件。说明 Spark作业主文件目前只支持存储在OSS中。
jars	Spark 作业依赖的 Jar 路径。
className	Java或者Scala程序入口类名称。Python不需要指定入口类。
args	请根据业务需求，填写使用Jar包时需要的参数。多个参数之间以英文逗号（,）分隔。
spark.kubernetes.driverEnv.ACCESS_KEY_ID	访问 Tablestore 的 Access Key ID。
spark.kubernetes.driverEnv.ACCESS_KEY_SECRET	访问 Tablestore 的 Access Key Secret 。
spark.hadoopRDD.ignoreEmptySplits	Spark 原生参数，用于控制是否忽略空分区。说明访问 Tablestore 实例必须将参数设置为 false。
conf其他参数	与开源Spark中的配置项基本一致，参数格式为key:value形式，多个参数之间以英文逗号（,）分隔。更多conf参数，请参见Conf配置参数。

单击立即执行。
待应用列表中目标应用的状态为已完成，您可以单击操作列的日志查看Table表的数据。