如何使用Spark DataFrame API开发一个流式作业消费LogService数据_开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心

本文简单介绍如何使用Spark DataFrame API开发一个流式作业消费LogService数据。

Spark Structured Streaming Scala访问LogHub

代码示例

## StructuredLoghubSample.Scala

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()

  }
}

说明

Maven pom文件可以参见aliyun-emapreduce-demo。

编译运行

## 编译命令
mvn clean package -DskipTests

## 编译完后，作业JAR包位于target目录下。

## 提交执行
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>

说明

作业资源需要根据实际数据规模和实际集群规模调整，如果集群规模太小，直接运行以上命令可能无法执行。

以下信息，请根据您实际环境替换：

x.x.x.StructuredLoghubSample：其中的x.x.x为您实际环境中StructuredLoghubSample类的包名。
xxx.jar：打包项目工程后的JAR包。
<starting-offsets>：开启offset位置，取值为earliest和latest。
<max-offsets-per-trigger>：每个触发周期内处理的消息或偏移量的最大数量。
<output-path>：输出数据的目录。例如，/loghub/data/。
<checkpoint-location>：checkpoint目录。例如，/loghub/checkpoint。
--jars：必须加上该参数，参数值为LogHub的Spark DataSource的JAR。如果不加上该参数，则会报Caused by: java.lang.ClassNotFoundException: loghub.DefaultSource。
- 针对Spark2，对应内容如下。
```
--jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark2-emrsdk/emr-datasources_shaded_2.11-2.3.1.jar
```
- 针对Spark3，对应内容如下。
```
--jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/emr-datasources_shaded_2.12-3.0.2.jar
```
说明
如果您的集群中没有以上目录，则使用/usr/lib/emrsdk-current/目录。

PySpark Structured Streaming访问LogHub

代码示例

from pyspark.sql import SparkSession

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

# 读取LogHub数据源。
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()


# 处理transform逻辑。
wordCounts = lines.groupBy("__logStore__").count()

# 处理Sink逻辑。
query = wordCounts \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

配置参数说明

参数	描述
endpoint	LogHub的endpoint。例如，cn-hangzhou-intranet.log.aliyuncs.com。
access.key.id	您阿里云账号的AccessKey ID。
access.key.secret	您阿里云账号的AccessKey Secret。
sls.project	LogStore名。
sls.store	LogService项目名。
startingoffsets	开启offset位置，取值为earliest和latest。

执行Python脚本

针对Spark2，对应内容如下。

spark-submit --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark2-emrsdk/emr-datasources_shaded_2.11-2.3.1.jar --master local loghub.py

针对Spark3，对应内容如下。

spark-submit --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/emr-datasources_shaded_2.12-3.0.2.jar --master local loghub.py

说明

如果您的集群中没有以上目录，则使用/usr/lib/emrsdk-current/目录。