Use Spark Streaming to consume data in real time

Sample code

The following example reads from an SLS Logstore, casts log values to strings, and writes output as Parquet files with a 30-second processing trigger.

object StructuredLoghubSample {
  def main(args: Array[String]) {
    if (args.length < 7) {
      System.err.println("Usage: StructuredLoghubSample <logService-project> " +
        "<logService-store> <access-key-id> <access-key-secret> <endpoint> " +
        "<starting-offsets> <max-offsets-per-trigger>[outputPath] [<checkpoint-location>]")
      System.exit(1)
    }

    val Array(project, logStore, accessKeyId, accessKeySecret, endpoint, startingOffsets, maxOffsetsPerTrigger, outputPath, _*) = args
    val checkpointLocation =
      if (args.length > 8) args(8) else "/tmp/temporary-" + UUID.randomUUID.toString

    val spark = SparkSession
      .builder
      .appName("StructuredLoghubSample")
      .getOrCreate()

    import spark.implicits._

    // Create DataSet representing the stream of input lines from loghub
    val lines = spark
      .readStream
      .format("loghub")
      .option("sls.project", project)
      .option("sls.store", logStore)
      .option("access.key.id", accessKeyId)
      .option("access.key.secret", accessKeySecret)
      .option("endpoint", endpoint)
      .option("startingoffsets", startingOffsets)
      .option("maxOffsetsPerTrigger", maxOffsetsPerTrigger)
      .load()
      .selectExpr("CAST(__value__ AS STRING)")
      .as[String]

    val query = lines.writeStream
      .format("parquet")
      .option("checkpointLocation", checkpointLocation)
      .option("path", outputPath)
      .outputMode("append")
      .trigger(Trigger.ProcessingTime(30000))
      .start()

    query.awaitTermination()

  }
}

For the Maven POM file, see aliyun-emapreduce-demo.

Build and submit

Step 1: Build the JAR.

mvn clean package -DskipTests

The compiled JAR is placed in the target/ directory.

Step 2: Submit the job.

spark-submit \
  --master --master yarn \
  --deploy-mode cluster \
  --executor-cores 2 \
  --executor-memory 1g \
  --driver-memory 1g \
  --num-executors 2 \
  --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark2-emrsdk/emr-datasources_shaded_2.11-2.3.1.jar \
  --class x.x.x.StructuredLoghubSample xxx.jar \
  <logService-project> <logService-store> <access-key-id> <access-key-secret> <endpoint> \
  <starting-offsets><max-offsets-per-trigger> <output-path> <checkpoint-location>

Replace the placeholders:

Adjust resource settings (--executor-cores, --executor-memory, --num-executors) based on your actual data volume and cluster capacity.

--jars path by Spark version:

If the path above does not exist in your cluster, use /usr/lib/emrsdk-current/ instead.

Sample code

The following example reads from an SLS Logstore and counts log entries by Logstore name.

The sample uses hardcoded placeholder credentials. In production, read credentials from environment variables to avoid exposing sensitive information in your code.

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("xx") \
    .getOrCreate()

# Read data from the LogHub data source
lines = spark \
    .readStream \
    .format("loghub") \
    .option("endpoint", "cn-hangzhou-intranet.log.aliyuncs.com") \
    .option("access.key.id", "LTAI----") \
    .option("access.key.secret", "DTi----") \
    .option("sls.project", "emr-test-hz-1") \
    .option("sls.store", "test1") \
    .option("startingoffsets", "earliest") \
    .load()

# Apply transformation
wordCounts = lines.groupBy("__logStore__").count()

# Write to console sink
query = wordCounts \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

Parameters

Run the script

Save the sample as loghub.py, then submit it with --jars pointing to the loghub DataSource JAR.

If the path above does not exist in your cluster, use /usr/lib/emrsdk-current/ instead.