訪問Tablestore資料來源

前提條件

• AnalyticDB for MySQL叢集的產品系列為企業版、基礎版或湖倉版。

• 已在AnalyticDB for MySQL叢集中建立Job型資源群組。

• • 如果是通過阿里雲帳號訪問，只需建立高許可權帳號。

  • 如果是通過RAM使用者訪問，需要建立高許可權帳號和普通帳號並且將RAM使用者綁定到普通帳號上。

• 已授權AnalyticDB for MySQL扮演AliyunADBSparkProcessingDataRole角色來訪問其他雲資源。

• AnalyticDB for MySQL叢集與TableStore執行個體位於同一地區。

操作步驟

1. 下載AnalyticDB for MySQLSpark訪問Tablestore依賴的Jar包。下載連結emr-tablestore-2.3.0-SNAPSHOT.jar和tablestore-5.10.3-jar-with-dependencies.jar。

2. 在pom.xml檔案的dependencies中添加依賴項。

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

3. 編寫如下樣本程式來訪問Tablestore，並進行編譯打包。本文產生的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 SparkOnTablestore {
  def main(args: Array[String]): Unit = {
    //Tablestore執行個體的VPC串連地址。
    val endpoint = args(0)
    // Tablestore的表名。
    val tableName = args(1)
    // Tablestore執行個體名稱。
    val instanceName = args(2)
    // 具備Tablestore讀寫權限的RAM使用者的AccessKey ID。
    val accessKeyId = sys.env.get("ACCESS_KEY_ID")
    //  具備Tablestore讀寫權限的RAM使用者的AccessKey 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.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()
    // 讀取資料。
    val df = spark.read.format("tablestore").options(options).load()
    df.show()
    // 寫入資料。
    df.write.format("tablestore").options(options).save()

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

4. 將步驟1和步驟3中的Jar包上傳至OSS。具體操作，請參見簡單上傳。

5. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊集群清單，然後單擊目的地組群ID。

6. 在左側導覽列，單擊作業開發 > Spark Jar開發。

7. 在編輯器視窗上方，選擇Job型資源群組和作業類型。本文以Batch類型為例。

8. 在編輯器中輸入以下作業內容。

{
  "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
  }
}

參數說明如下。

7. 單擊立即执行。

8. 待應用列表中目標應用的狀態為已完成，您可以單擊操作列的日志查看Tablestore表的資料。