Consume DataHub data with Spark Streaming and Spark Structured Streaming - E-MapReduce

E-MapReduce (EMR) Hadoop clusters support two Spark APIs for consuming DataHub data: Spark Streaming and Spark Structured Streaming. This topic shows how to set up each approach with working code examples.

Choose an API

	Spark Streaming	Spark Structured Streaming
API generation	Older (DStream-based)	Newer (DataFrame-based)
Offset management	DataHub manages offsets via a subscription (`subId`)	Spark manages offsets internally — no `subId` needed
Shard control	Consume a single shard or all shards	Reads all shards automatically
Use when	You need shard-level control or are maintaining existing code	Starting a new pipeline or prefer the DataFrame API

Use Spark Streaming to consume DataHub data

Prerequisites

Before you begin, ensure that you have:

An EMR Hadoop cluster
A DataHub topic with at least one shard
A DataHub subscription ID (subId) — see Create a subscription

Consume data from a specific shard

Pass a shardId to DatahubUtils.createStream to read from one shard:

datahubStream = DatahubUtils.createStream(
  ssc,
  project,        // DataHub project name
  topic,          // DataHub topic name
  subId,          // DataHub subscription ID
  accessKeyId,
  accessKeySecret,
  endpoint,       // DataHub endpoint
  shardId,        // ID of the shard to consume
  read,           // Function to process each RecordEntry
  StorageLevel.MEMORY_AND_DISK)

datahubStream.foreachRDD(rdd => println(rdd.count()))

// Read data from the first field in RecordEntry
def read(record: RecordEntry): String = {
  record.getString(0)
}

Consume data from all shards

Omit the shardId argument to consume from all shards in the topic:

datahubStream = DatahubUtils.createStream(
  ssc,
  project,        // DataHub project name
  topic,          // DataHub topic name
  subId,          // DataHub subscription ID
  accessKeyId,
  accessKeySecret,
  endpoint,       // DataHub endpoint
  read,           // Function to process each RecordEntry
  StorageLevel.MEMORY_AND_DISK)

datahubStream.foreachRDD(rdd => println(rdd.count()))

// Read data from the first field in RecordEntry
def read(record: RecordEntry): String = {
  record.getString(0)
}

Note

For the complete sample code, see SparkDatahubDemo.scala.

Use Spark Structured Streaming to consume DataHub data

Spark Structured Streaming treats DataHub as a streaming source and manages offsets internally. You do not need a DataHub subscription ID (subId).

Add the dependency

Spark 2

Add the following Maven dependency:

<dependency>
  <groupId>com.aliyun.emr</groupId>
  <artifactId>emr-datahub_2.11</artifactId>
  <version>2.0.0</version>
</dependency>

Spark 3

Download emr-datasources_shaded_***.jar from the cluster directory /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/ and add it as a dependency.

Note

If that directory does not exist in your cluster, use /usr/lib/emrsdk-current/ instead. Replace emr-datasources_shaded_***.jar with the actual filename in your cluster directory.

Read and process DataHub data

The following example creates a read stream from DataHub, groups records by key, and writes counts to the console:

val spark = SparkSession
  .builder()
  .appName("test datahub")
  .getOrCreate()

// Create a read stream from DataHub
val datahubRows = spark
  .readStream
  .format("datahub")
  .option("access.key.id", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
  .option("access.key.secret", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
  .option("endpoint", "https://dh-cn-hangzhou.aliyuncs.com")
  .option("project", "project_test")
  .option("topic", "topic_test")
  .option("startingoffsets", "earliest")
  .load

// The schema has two fields: key and value
datahubRows.printSchema()
println("print schema" + datahubRows.schema.toString())
val df = datahubRows.groupBy("key").count()

// Write aggregated results to the console
val query = df
  .writeStream
  .format("console")
  .outputMode("complete")
  .start()

query.awaitTermination(100000)
spark.close()

The core process:

Create a read stream to load data from DataHub into a DataFrame.
Apply processing logic to the DataFrame.
Create a write stream to output the results.

Important

Configure environment variables before running this code. For details, see Configure environment variables.

Parameters

Required

Parameter	Description
`access.key.id`	AccessKey ID of the Alibaba Cloud account used to create the DataHub project
`access.key.secret`	AccessKey secret of the Alibaba Cloud account used to create the DataHub project
`endpoint`	Endpoint of the DataHub API. View the endpoint on the DataHub page.
`project`	Name of the DataHub project
`topic`	Name of the DataHub topic

Optional

Parameter	Description
`startingoffsets`	Offset from which data consumption starts. Valid values: `latest`, `earliest`, or a JSON string specifying per-shard offsets (see below).
`endingoffsets`	Offset at which data consumption ends. Valid values: `latest` or a JSON string specifying per-shard offsets (see below).
`decimal.precision`	Required if the topic contains a field of the DECIMAL type.
`decimal.scale`	Required if the topic contains a field of the DECIMAL type.

JSON format for per-shard offsets

Use the following format to specify offsets for individual shards:

{
  "<project>#<topic>": {
    "<shardId>": "<offset>"
  }
}

Example:

{
  "project_test#topic_test": {
    "0": "100"
  }
}