Run a Structured Streaming task in EMR Serverless Spark to read from and write to ApsaraMQ for Kafka. This guide walks you through uploading the required JAR files, creating a streaming task, and verifying results using logs or the Kafka console.
Prerequisites
Before you begin, make sure you have:
-
A Serverless Spark workspace. For more information, see Create a workspace.
-
An ApsaraMQ for Kafka instance. This guide uses a cloud-based instance. For more information, see Purchase and deploy an instance.
Step 1: Upload Kafka-related JARs to OSS
EMR Serverless Spark engine versions esr-2.8.0, esr-3.4.0, esr-4.4.0, and later include built-in Kafka JAR packages. If you use one of these versions, skip this step.
For other engine versions, download and upload the required JARs manually:
-
Download the JAR package for your Spark version:
-
Spark 3.5: kafka-spark35-jars.zip
-
Spark 3.4: kafka-spark34-jars.zip
-
Spark 3.3: kafka-spark33-jars.zip
-
-
Decompress the file and upload all JAR packages to Object Storage Service (OSS). The following example uses kafka-spark35-jars.zip:
You can also upload files using the OSS console or other methods.
hadoop fs -put /root/spark-sql-kafka-0-10_2.12-3.5.3.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/ hadoop fs -put /root/kafka-clients-3.4.1.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/ hadoop fs -put /root/spark-token-provider-kafka-0-10_2.12-3.5.3.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/ hadoop fs -put /root/commons-pool2-2.11.1.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/
Step 2: Create a network connection
EMR Serverless Spark requires network connectivity to your ApsaraMQ for Kafka instance. For setup instructions, see Network connectivity between EMR Serverless Spark and other VPCs.
Step 3: Prepare the Scala code
The following examples show how to read from and write to Kafka using Spark Structured Streaming. Both examples accept the Kafka endpoint and topic as runtime arguments.
To use a prebuilt JAR, download SparkExample-1.0-SNAPSHOT-jar-with-dependencies.jar and skip to Step 4. To package the code yourself, see Appendix: pom.xml for the Maven configuration. For additional examples, see the Spark Kafka integration documentation.
Read from Kafka
This example subscribes to a Kafka topic and prints messages to the console as [key, value] pairs. The CAST expressions convert the binary key and value columns to strings.
object StreamingReadKafka {
def main(args: Array[String]): Unit = {
import org.apache.spark.sql.SparkSession
val servers = args(0)
val topic = args(1)
val spark = SparkSession.builder()
.appName("test read kafka")
.getOrCreate()
import spark.implicits._
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", servers)
.option("subscribe", topic)
.load()
val query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.as[(String, String)]
.writeStream
.format("console")
.start()
query.awaitTermination()
}
}Write to Kafka
This example generates records from the built-in Spark rate source at one record per second, uses the value column as both the key and value, and writes to Kafka every 10 seconds.
object StreamingWriteKafka {
def main(args: Array[String]): Unit = {
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger
val servers = args(0)
val topic = args(1)
val checkpointDir = args(2)
val spark = SparkSession.builder()
.appName("test write kafka")
.getOrCreate()
val df = spark.readStream
.format("rate")
.option("rowsPerSecond", "1")
.load()
.withColumn("key", col("value"))
.withColumn("value", col("value"))
val query = df
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", servers)
.option("topic", topic)
.option("checkpointLocation", checkpointDir)
.trigger(Trigger.ProcessingTime("10 seconds"))
.start()
query.awaitTermination()
}
}Step 4: Upload the JAR package
-
Log on to the E-MapReduce console.
-
In the left navigation pane, choose EMR Serverless > Spark.
-
On the Spark page, click the target workspace.
-
In the left navigation pane, click Artifacts.
-
On the Artifacts page, click Upload File.
-
In the Upload File dialog box, click the upload area or drag the JAR package into it.
Step 5: Create and run a streaming task
-
On the EMR Serverless Spark page, click Development in the left navigation pane.
-
On the Development tab, click the New icon
. -
In the dialog box, enter a Name, select Application(Streaming) as the type, and click OK.
-
In the upper-right corner, select the target queue. For more information, see Manage resource queues.
-
Configure the following parameters and leave all others at their default values. Then click Publish. Execution parameters differ depending on whether you are reading or writing: If you uploaded Kafka JARs in Step 1, add the following to Spark Configurations:
Parameter Description Main JAR Resource Select the JAR package uploaded in Step 4. In this example: SparkExample-1.0-SNAPSHOT-jar-with-dependencies.jar.Engine Version Select a Spark engine version. In this example: esr-4.3.0.Main Class The main class to run. For reading: org.example.StreamingReadKafka. For writing:org.example.StreamingWriteKafka.Execution Parameters Arguments passed to the main class, separated by spaces. See the table below. Network Connection Select the network connection created in Step 2. Spark Configurations Use spark.emr.serverless.user.defined.jarsto specify Kafka JARs (if required by your engine version). Separate multiple paths with commas.Scenario Parameters (in order) Read from Kafka 1. Kafka endpoint (e.g., alikafka-serverless-cn-xxxxxx-1000-vpc.alikafka.aliyuncs.com:9092) 2. Topic name (e.g.,test)Write to Kafka 1. Kafka endpoint 2. Topic name 3. Checkpoint path (e.g., oss://<YOUR_BUCKET_PATH>/)spark.emr.serverless.user.defined.jars oss://<YOUR_BUCKET_PATH>/commons-pool2-2.11.1.jar,oss://<YOUR_BUCKET_PATH>/kafka-clients-3.4.1.jar,oss://<YOUR_BUCKET_PATH>/spark-sql-kafka-0-10_2.12-3.5.3.jar,oss://<YOUR_BUCKET_PATH>/spark-token-provider-kafka-0-10_2.12-3.5.3.jar -
After publishing, click Go To O&M, then click Start.
Step 6: Verify the results
Read from Kafka
-
On the Workflows > Streaming Deployments tab, click the target task.
-
Open the Log Exploration
Write to Kafka
-
Log on to the Kafka console and navigate to the instance page.
-
On the Message Query tab, click Query
Appendix: pom.xml
The following Maven configuration builds the example JAR with all dependencies included.
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.5.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.5.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.5.2</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>com.example.Main</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>