AnalyticDB:Access Tablestore

Step 1: Download the required JAR packages

Download the following JAR packages:

• emr-tablestore-2.3.0-SNAPSHOT.jar

• tablestore-5.10.3-jar-with-dependencies.jar

Step 2: Add the Maven dependency

Add the following dependency to your pom.xml:

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

Step 3: Write and package the Spark program

Write a Spark program that reads from and writes to Tablestore, then package it as a JAR file. The following example reads all rows from a Tablestore table and writes them back. The output JAR in this example is named spark-tablestore.jar.

All examples use the tablestore format with the options map to configure the endpoint, credentials, table name, and data catalog (column schema).

package com.aliyun.spark

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

object SparkOnTablestore {
  def main(args: Array[String]): Unit = {
    // VPC endpoint of the Tablestore instance
    val endpoint = args(0)
    // Name of the Tablestore table
    val tableName = args(1)
    // Name of the Tablestore instance
    val instanceName = args(2)
    // AccessKey ID of the RAM user — read from environment variables
    val accessKeyId = sys.env.get("ACCESS_KEY_ID")
    // AccessKey secret of the RAM user — read from environment variables
    val accessKeySecret = sys.env.get("ACCESS_KEY_SECRET")
    // Data catalog: maps Spark SQL columns to Tablestore column names and types
    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.get)
    options.put("access.key.secret", accessKeySecret.get)
    options.put("table.name", tableName)
    options.put("instance.name", instanceName)
    options.put("catalog", dataCatalog)

    val spark = SparkSession.builder().getOrCreate()
    // Read data from Tablestore
    val df = spark.read.format("tablestore").options(options).load()
    df.show()
    // Write data back to Tablestore
    df.write.format("tablestore").options(options).save()

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

Step 4: Upload JAR packages to OSS

Upload the JAR packages from Steps 1 and 3 to Object Storage Service (OSS). See Simple upload.

Step 5: Open Spark JAR development in the console

1. Log on to the AnalyticDB for MySQL console. In the upper-left corner, select a region.

2. In the left-side navigation pane, click ClustersData Lakehouse Edition. On the Data Lakehouse Edition tab, find the cluster and click its ID.

3. In the left-side navigation pane, choose Job Development > Spark JAR Development.

Step 6: Configure and run the Spark job

1. Select a job resource group and set the job type to Batch.

2. Enter the following configuration in the Spark editor. Replace the placeholders with your actual values.

   {
     "args": [
       "https://i01164l****.<region_id>.vpc.tablestore.aliyuncs.com",
       "spark_test",
       "i01164l****"
     ],
     "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-tablestore",
     "className": "com.aliyun.spark.SparkOnTablestore",
     "conf": {
       "spark.driver.resourceSpec": "medium",
       "spark.executor.instances": 2,
       "spark.executor.resourceSpec": "medium",
       "spark.kubernetes.driverEnv.ACCESS_KEY_ID": "LTAI****************",
       "spark.kubernetes.driverEnv.ACCESS_KEY_SECRET": "yourAccessKeySecret",
       "spark.hadoopRDD.ignoreEmptySplits": false
     }
   }

   The following table describes the key parameters.

   Parameter	Description	Example
   args	Arguments passed to the JAR program, in the order expected by main(args): Tablestore VPC endpoint, table name, and instance name	"https://i01164l****.<region_id>.vpc.tablestore.aliyuncs.com"
   file	OSS path of the application JAR (spark-tablestore.jar)	"oss://my-bucket/spark-tablestore.jar"
   jars	OSS paths of the dependency JARs required by the Spark job	"oss://my-bucket/emr-tablestore-2.3.0-SNAPSHOT.jar"
   name	Display name of the Spark job	"spark-on-tablestore"
   className	Entry class of the Scala or Java program. Not required for Python applications	"com.aliyun.spark.SparkOnTablestore"
   spark.kubernetes.driverEnv.ACCESS_KEY_ID	AccessKey ID of the RAM user used to access Tablestore, injected as an environment variable into the Spark driver	"LTAI5tXxx"
   spark.kubernetes.driverEnv.ACCESS_KEY_SECRET	AccessKey secret of the RAM user used to access Tablestore, injected as an environment variable into the Spark driver	"yourAccessKeySecret"
   spark.hadoopRDD.ignoreEmptySplits	Whether to skip empty partitions. Must be set to `false` when accessing Tablestore.	false
   conf	Spark configuration in key:value format, compatible with Apache Spark parameters. Separate multiple parameters with commas (,). See Spark application configuration parameters	—

3. Click Run Now.

Verify the result

After submitting the job, go to the Applications tab and wait for the Spark application status to change to Completed. Then click Logs in the Actions column to view the Tablestore table data printed by df.show().