AnalyticDB:Access data in Tablestore

Prerequisites

• The product series of the AnalyticDB for MySQL cluster is Enterprise, Basic, or Data Lakehouse Edition.

• A Job-type resource group has been created in the AnalyticDB for MySQL cluster.

• • If you access Tablestore by using an Alibaba Cloud account, you only need to create a privileged account.

  • If you access Tablestore by using a RAM user, you need to create a privileged account and a standard account, and bind the RAM user to the standard account.

• AnalyticDB for MySQL has been authorized to assume the AliyunADBSparkProcessingDataRole role to access other cloud resources.

• The AnalyticDB for MySQL cluster and the Tablestore instance are in the same region.

Procedure

1. Download the JAR packages required for Spark in AnalyticDB for MySQL to access Tablestore: emr-tablestore-2.2.0.jar and tablestore-5.10.3-jar-with-dependencies.jar.

2. Add the following dependency to the dependencies block in your pom.xml file.

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

3. Write a sample program to access Tablestore, and then compile and package it. In this example, the output JAR package is named spark-tablestore.jar. Here is the sample code:

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 = {
    // The VPC endpoint of the Tablestore instance.
    val endpoint = args(0)
    // The name of the Tablestore table.
    val tableName = args(1)
    // The name of the Tablestore instance.
    val instanceName = args(2)
    // The AccessKey ID of the RAM user that has the permissions to read from and write to Tablestore.
    val accessKeyId = sys.env.get("ACCESS_KEY_ID")
    // The AccessKey Secret of the RAM user that has the permissions to read from and write to Tablestore.
    val accessKeySecret = sys.env.get("ACCESS_KEY_SECRET")
    // The data catalog of the Tablestore table.
    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.
    val df = spark.read.format("tablestore").options(options).load()
    df.show()
    // Write data.
    df.write.format("tablestore").options(options).save()

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

4. Upload the JAR packages from Step 1 and Step 3 to OSS. For more information, see Simple upload.

5. Log on to the AnalyticDB for MySQL console. In the upper-left corner of the console, select a region. In the left-side navigation pane, click Clusters. Find the cluster that you want to manage and click the cluster ID.

6. In the left-side navigation pane, click Job Development > Spark Jar Development.

7. Above the editor, select a job resource group and a job type. This example uses the Batch type.

8. In the editor, enter the following job configuration.

{
  "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.2.0.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 parameters.

7. Click Immediately.

8. When the application's status in the Workspaces list changes to Completed, click Log in the Actions column to view the Tablestore table data.