How to use Spark to access ApsaraMQ for Kafka over an elastic network interface - AnalyticDB

This topic describes how to use Spark in AnalyticDB for MySQL to access data from ApsaraMQ for Kafka over an elastic network interface (ENI).

Prerequisites

An AnalyticDB for MySQL Enterprise Edition, Basic Edition, or Data Lakehouse Edition cluster is created.
A job resource group is created for the AnalyticDB for MySQL cluster.
A database account is created for the AnalyticDB for MySQL cluster.
- If you use an Alibaba Cloud account, you need to only create a privileged account.
- If you use a Resource Access Management (RAM) user, you must create a privileged account and a standard account and associate the standard account with the RAM user.
Ensure the ApsaraMQ for Kafka instance and the AnalyticDB for MySQL cluster are in the same region. For more information, see Internet and VPC connections.
Create a topic and a consumer group in the ApsaraMQ for Kafka instance. For more information, see Create resources.
AnalyticDB for MySQL (OSS) must be activated and a bucket must be created in the same region as your AnalyticDB for MySQL cluster. For more information, see Activate OSS and Create buckets.

Gather required information

In the ApsaraMQ for Kafka console, go to the Instance Details page to obtain the vSwitch ID of your Kafka instance.
In the Elastic Compute Service (ECS) console, go to the Security Group page and search for the Kafka instance ID to get the security group ID.
In the ApsaraMQ for Kafka console, go to the Manage Whitelist page and verify that the whitelist for the Kafka instance contains the CIDR block of the vSwitch.

Procedure

Download the JAR packages that match the versions of your Kafka and AnalyticDB for MySQL Spark instances. For download links, see Kafka-clients and Spark-sql-kafka-0-10.

Add the dependencies to the dependencies section of your pom.xml file.

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

Write a Spark Streaming sample program to read messages from Kafka, and then compile and package the program. In this example, the generated JAR package 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 endpoints of the Kafka instance.
      .option("kafka.bootstrap.servers", bootstrapServers)
     // The name of the Kafka topic.
      .option("subscribe", topicName)
     // The ID of the Kafka consumer group.
      .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()

  }
}

Upload the downloaded JAR packages and the Spark Streaming sample program to OSS. For more information, see Upload objects.
Go to the Spark development editor.
1. 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.
2. In the left-side navigation pane, click Job Development > Spark JAR Development.
Above the editor window, select a job resource group and a Spark job type. This example uses the Batch type.

In the editor, run the following job configuration.

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

Parameter	Description
`args`	The arguments passed to the Spark job. Separate multiple arguments with a comma (,).
`file`	The storage location of the main file for the Spark job. The main file is the JAR package that contains the entry class or the entry Python file. Note The main file of a Spark job must be stored in OSS.
`jars`	The dependency JAR packages for the Spark job. Separate multiple JAR packages with a comma (,).
`name`	The name of the Spark job.
`className`	The entry class for the Java or Scala program. This parameter is not required for Python programs.
`spark.adb.eni.enabled`	Specifies whether to enable access over an ENI. You must enable this feature when you use Spark in an Enterprise Edition, Basic Edition, or Data Lakehouse Edition cluster to access a Kafka data source.
`spark.adb.eni.vswitchId`	The vSwitch ID that you obtained in the "Gather required information" step.
`spark.adb.eni.securityGroupId`	The security group ID that you obtained in the "Gather required information" step.
Other `conf` parameters	These parameters are mostly consistent with open source Spark configurations. Use the `key:value` format. Separate multiple parameters with a comma (,). For parameters that differ from open source Spark or are specific to AnalyticDB for MySQL, see Spark application configuration parameters.

Click Run Now.