Apache Spark Streaming计算结果从HDFS导出至S3、RDS的精准触发方案问询

针对你在EMR上运行Spark Streaming小时窗口作业，需要在每次计算完成后自动触发S3Distcp和Sqoop导出的需求，我整理了几个生产环境中常用的精准解决方案：

方案一：在Spark作业内部直接触发导出（最精准） #

既然Spark本身清楚窗口计算完成的时机，直接在窗口结果的处理逻辑后嵌入导出命令，是最能保证“计算完成即触发”的方式。

对于传统DStream API #

假设你已经将窗口计算结果写入HDFS的唯一路径（用窗口时间戳命名，避免冲突），可以在foreachRDD里执行外部命令：

import org.apache.spark.streaming._

val ssc = new StreamingContext(sparkConf, Seconds(10))
val inputStream = ssc.socketTextStream("数据源地址")

// 定义1小时滚动窗口
val windowedStream = inputStream.window(Minutes(60), Minutes(60))

windowedStream.foreachRDD { (rdd, windowTime) =>
  // 先执行核心计算，将结果写入HDFS唯一路径
  val outputDir = s"/hourly-results/${windowTime.milliseconds}"
  rdd.saveAsTextFile(outputDir)

  // 触发S3Distcp导出到S3
  val s3DistcpCmd = s"/usr/bin/s3-dist-cp --src $outputDir --dest s3://你的目标Bucket/hourly-data/${windowTime.milliseconds}"
  val s3Process = Runtime.getRuntime.exec(s3DistcpCmd)
  val s3ExitCode = s3Process.waitFor()
  if (s3ExitCode != 0) {
    // 可根据需求重试、报警或终止作业
    throw new RuntimeException(s"S3Distcp导出失败，退出码：$s3ExitCode，窗口时间：$windowTime")
  }

  // 触发Sqoop导出到RDS（添加update-key保证幂等，避免重复写入）
  val sqoopCmd = 
    s"/usr/bin/sqoop export " +
    s"--connect jdbc:mysql://你的RDS地址:3306/目标库 " +
    s"--username 用户名 " +
    s"--password 密码 " +
    s"--table 目标表 " +
    s"--export-dir $outputDir " +
    s"--input-fields-terminated-by ',' " +
    s"--update-key 主键字段"
  val sqoopProcess = Runtime.getRuntime.exec(sqoopCmd)
  val sqoopExitCode = sqoopProcess.waitFor()
  if (sqoopExitCode != 0) {
    throw new RuntimeException(s"Sqoop导出失败，退出码：$sqoopExitCode，窗口时间：$windowTime")
  }
}

ssc.start()
ssc.awaitTermination()

注意：EMR环境下s3-dist-cp和sqoop的默认路径是/usr/bin/，用绝对路径可避免环境变量问题；另外，Runtime.exec会占用Executor资源，如果作业资源紧张，建议用方案二解耦。

对于Structured Streaming（推荐） #

如果用的是更现代的Structured Streaming，可用foreachBatch回调实现相同逻辑：

import org.apache.spark.sql.streaming.Trigger

val df = spark.readStream.format("kafka").load()
// 你的窗口计算逻辑...

val query = df.writeStream
  .trigger(Trigger.ProcessingTime("1 hour"))
  .foreachBatch { (batchDF, batchId) =>
    val outputDir = s"/hourly-results/batch-$batchId"
    // 保存计算结果到HDFS
    batchDF.write.parquet(outputDir)

    // 执行S3Distcp和Sqoop命令，逻辑同DStream示例
    // ...
  }
  .option("checkpointLocation", "/spark-checkpoint")
  .start()

query.awaitTermination()

方案二：用EMR Step解耦导出操作 #

如果不想让导出逻辑占用Spark作业资源，可以在计算完成后，通过EMR API提交独立的Shell Step执行导出。这样导出操作与Spark作业完全解耦，失败后可单独重试Step。

示例代码（Scala中用AWS SDK调用EMR API）：

import software.amazon.awssdk.services.emr.EmrClient
import software.amazon.awssdk.services.emr.model.{AddJobFlowStepsRequest, HadoopJarStepConfig, StepConfig}

// 从EMR环境变量获取当前集群ID
val clusterId = sys.env.getOrElse("EMR_CLUSTER_ID", throw new RuntimeException("未找到EMR_CLUSTER_ID环境变量"))

val emrClient = EmrClient.create()

// 定义导出Step：执行s3-dist-cp和sqoop命令
val exportStep = StepConfig.builder()
  .name(s"导出批次$batchId的小时结果")
  .hadoopJarStep(HadoopJarStepConfig.builder()
    .jar("command-runner.jar") // EMR自带的命令运行器，用于执行Shell命令
    .args(
      "bash", "-c",
      s"""
         |/usr/bin/s3-dist-cp --src /hourly-results/batch-$batchId --dest s3://你的目标Bucket/...;
         |if [ $? -ne 0 ]; then exit 1; fi;
         |/usr/bin/sqoop export --connect ... --export-dir /hourly-results/batch-$batchId ...;
         |""".stripMargin
    )
    .build())
  .actionOnFailure("CONTINUE") // 导出失败后继续集群运行，也可设为TERMINATE_JOB_FLOW
  .build()

// 提交Step到当前EMR集群
val request = AddJobFlowStepsRequest.builder()
  .jobFlowId(clusterId)
  .steps(exportStep)
  .build()

emrClient.addJobFlowSteps(request)

注意：需要给EMR的EC2实例角色添加elasticmapreduce:AddJobFlowSteps权限，否则无法提交Step。

关键注意事项 #

• 幂等性优先：无论用哪种方案，都要保证导出操作是幂等的。比如S3路径用唯一的窗口时间/批次ID，Sqoop用--update-key实现upsert，避免重复导出导致数据重复。
• 错误处理与报警：添加重试机制（比如S3Distcp失败重试2次），或者将错误信息发送到AWS SNS/CloudWatch告警，方便及时排查问题。
• 资源隔离：方案一中的Runtime.exec会占用Executor的CPU和内存，如果Spark作业资源紧张，优先选方案二，让导出Step在集群的Task节点上单独运行。
• 权限配置：确保EMR实例角色有S3读写权限、RDS写入权限（如果用Sqoop），以及EMR API权限（如果用方案二）。

内容的提问来源于stack exchange，提问作者Laurynas Stašys