訪問Kafka資料來源

本文介紹如何使用AnalyticDB for MySQLSpark通過ENI網路訪問訊息佇列Kafka版。

前提條件

AnalyticDB for MySQL叢集的產品系列為企業版、基礎版或湖倉版。
已在AnalyticDB for MySQL叢集中建立Job型資源群組。
已建立AnalyticDB for MySQL叢集的資料庫帳號。
- 如果是通過阿里雲帳號訪問，只需建立高許可權帳號。
- 如果是通過RAM使用者訪問，需要建立高許可權帳號和普通帳號並且將RAM使用者綁定到普通帳號上。
Kafka執行個體與AnalyticDB for MySQL叢集位於同一地區。具體操作，請參見公網和VPC接入。
已在Kafka執行個體中建立Topic和Group。具體操作，請參見建立資源。
已開通OSS服務，並建立與AnalyticDB for MySQL叢集位於相同地區的儲存空間。具體操作，請參見開通OSS服務和建立儲存空間。

準備工作

在雲訊息佇列 Kafka 版控制台的實例詳情頁面，擷取Kafka執行個體的交換器ID。
在ECS管理主控台的安全組頁面，搜尋Kafka執行個體ID來擷取安全性群組ID。
在雲訊息佇列 Kafka 版控制台的白名單管理頁面，查看Kafka執行個體的白名單是否為交換器ID的網段。

操作步驟

分別下載與Kafka和AnalyticDB for MySQL Spark執行個體版本對應的JAR包。下載連結，請參見Kafka-clients和Spark-sql-kafka-0-10。

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

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql-kafka-0-10 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
    <version>3.2.2</version>
    <scope>test</scope>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.1</version>
</dependency>

編寫Spark Streaming樣本程式來讀取Kafka中的訊息，並進行編譯打包。本文產生的JAR包名稱為spark-example.jar。

package com.aliyun.spark.streaming

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object SparkKafka {
  def main(args: Array[String]): Unit = {
    if(args.length < 3){
      System.err.println(
        """
          |args0: groupId
          |args1: topicName
          |args2: bootstrapServers
          |""".stripMargin)
      System.exit(1)
    }
    val groupId = args(0)
    val topicName = args(1)
    val bootstrapServers = args(2)


    val sparkConf: SparkConf = new SparkConf()
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .setAppName("SparkKafkaSub")
    sparkConf.registerKryoClasses(Array(classOf[ConsumerRecord[_,_]]))

    val sparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()

    val df = sparkSession
      .readStream
      .format("kafka")
     //Kafka執行個體的網域名稱存取點。
      .option("kafka.bootstrap.servers", alikafka-pre-cn-x0r34a20****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20****-3-vpc.alikafka.aliyuncs.com:9092)
     //Kafka執行個體的Topic名稱。
      .option("subscribe", kafka_test)
     //Kafka執行個體的Group ID。
      .option("group.id", kafka_groupId)
      .load()

    val query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
      .writeStream
      .outputMode("append")
      .format("console")
      .start()
    query.awaitTermination()

  }
}

將下載的JAR包和Spark Streaming樣本程式上傳至OSS。具體操作，請參見上傳檔案。
進入Spark開發編輯器。
1. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊集群清單，然後單擊目的地組群ID。
2. 在左側導覽列，單擊作業開發 > Spark Jar 開發。
在編輯器視窗上方，選擇Job型資源群組和Spark作業類型。本文以Batch類型為例。

在編輯器中執行以下作業內容。

{
    "args": [
      "kafka_groupId",
      "kafka_test",
      "alikafka-pre-cn-x0r34a20****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-x0r34a20****-3-vpc.alikafka.aliyuncs.com:9092"
    ],
    "file": "oss://<bucket_name>/spark-example.jar",
    "jars": "oss://<bucket_name>/kafka-clients-2.8.1.jar,oss://<bucket_name>/spark-sql-kafka-0-10_2.12-3.2.0.jar",
    "name": "Kafka Example",
    "className": "com.aliyun.spark.streaming.SparkKafka",
    "conf": {
        "spark.driver.resourceSpec": "small",
        "spark.executor.instances": 1,
        "spark.executor.resourceSpec": "small",
        "spark.adb.eni.enabled": "true",
        "spark.adb.eni.vswitchId": "vsw-bp17jqw3lrrobn6y****",
        "spark.adb.eni.securityGroupId": "sg-bp163uxgt4zandx****"
    }
}

參數說明如下。

參數名稱	參數說明
`args`	Spark作業傳入的參數，多個參數之間以英文逗號（,）分隔。
`file`	Spark作業主檔案的儲存位置。主檔案是入口類所在的JAR包或者Python的入口執行檔案。說明 Spark作業主檔案目前只支援儲存在OSS中。
`jars`	Spark作業依賴的JAR包，多個JAR包之間以英文逗號（,）分隔。
`name`	Spark作業名稱。
`className`	Java或者Scala程式入口類。Python不需要指定入口類。
`spark.adb.eni.enabled`	是否開啟ENI訪問。使用企業版、基礎版或湖倉版Spark訪問kafka資料來源時，需要開啟ENI訪問。
`spark.adb.eni.vswitchId`	準備工作中擷取的交換器ID。
`spark.adb.eni.securityGroupId`	準備工作中擷取的安全性群組ID。
`conf`其他參數	與開源Spark中的配置項基本一致，參數格式為`key:value`形式，多個參數之間以英文逗號（,）分隔。與開源Spark用法不一致的配置參數及AnalyticDB for MySQL特有的配置參數，請參見Spark應用配置參數說明。

單擊運行。