AnalyticDB:Access ApsaraMQ for Kafka

Step 1: Add dependencies

Add the following dependencies to your pom.xml. The Spark-Kafka connector version must match the Spark runtime version of your cluster.

<!-- 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>

For other versions, see kafka-clients and spark-sql-kafka-0-10 on Maven Repository.

Step 2: Write and package the Spark Streaming program

Write a Spark Streaming program that reads from ApsaraMQ for Kafka. The following sample code reads messages from a Kafka topic and prints them to the console. After packaging, the output JAR is named 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")
      // The endpoint of the Kafka instance.
      .option("kafka.bootstrap.servers", bootstrapServers)
      // The name of the topic to consume.
      .option("subscribe", topicName)
      // The consumer group ID.
      .option("group.id", groupId)
      .load()

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

Step 3: Upload JAR packages to OSS

Upload the following files to your OSS bucket. For upload instructions, see Upload objects.

• The downloaded kafka-clients-2.8.1.jar and spark-sql-kafka-0-10_2.12-3.2.0.jar

• The packaged spark-example.jar

Step 4: Submit the Spark job

1. Log on to the AnalyticDB for MySQL console. Select a region, then click Clusters in the left-side navigation pane.

2. On the Data Lakehouse Edition tab, find the cluster and click the cluster ID.

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

4. Select a job resource group and set the job type. This example uses the batch type.

5. Enter the following configuration in the Spark editor, replacing the placeholder values with your actual vSwitch ID and security group ID:

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

   The following table describes the key parameters. For all supported Spark configuration parameters, see Spark application configuration parameters.

   Parameter	Description
   args	The arguments passed to the Spark job. Separate multiple arguments with commas (,).
   file	The path of the main file of the Spark job. The main file can be a JAR package that contains the entry class or an executable file that serves as the entry point for the Python program. Note The main file of a Spark job must be stored in OSS.
   jars	The JAR packages that are required for a Spark job. Separate multiple JAR packages with commas (,).
   name	The name of the Spark job.
   className	The entry class of the Java or Scala program. This parameter is not required for a Python program.
   spark.adb.eni.enabled	Set to "true" to enable ENI. Required when accessing ApsaraMQ for Kafka from Enterprise Edition, Basic Edition, or Data Lakehouse Edition Spark.
   spark.adb.eni.vswitchId	The vSwitch ID of the ApsaraMQ for Kafka instance. Obtained in Gather network information.
   spark.adb.eni.securityGroupId	The ID of the security group associated with the ApsaraMQ for Kafka instance. Obtained in Gather network information.
   conf	The configuration parameters that are required for the Spark job, which are similar to those of Apache Spark. The parameters must be in the key:value format. Separate multiple parameters with commas (,). For more information about the configuration parameters that are different from those of Apache Spark or the configuration parameters specific to AnalyticDB for MySQL, see Spark application configuration parameters.

6. Click Run Now.