如何在Scala开发的AWS Glue作业中获取工作流运行时属性

以下是与你提供的Python代码功能完全对等的Scala版本，可直接在AWS Glue托管环境中运行：

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import org.apache.spark.SparkContext
import com.amazonaws.services.glue.AWSGlueClientBuilder

object WorkflowPropDemo {
  def main(args: Array[String]): Unit = {
    val sc = SparkContext.getOrCreate()
    val glueContext = GlueContext(sc)

    // 解析系统入参，等价Python的getResolvedOptions
    val glueArgs = GlueArgParser.getResolvedOptions(
      args,
      Array("JOB_NAME", "WORKFLOW_NAME", "WORKFLOW_RUN_ID")
    )

    // 初始化Glue服务客户端
    val glueClient = AWSGlueClientBuilder.defaultClient()

    val workflowName = glueArgs("WORKFLOW_NAME")
    val workflowRunId = glueArgs("WORKFLOW_RUN_ID")

    // 请求工作流运行属性
    val runPropertiesReq = new com.amazonaws.services.glue.model.GetWorkflowRunPropertiesRequest()
      .withName(workflowName)
      .withRunId(workflowRunId)
    val workflowParams = glueClient.getWorkflowRunProperties(runPropertiesReq).getRunProperties

    // 提取自定义属性
    val targetDatabase = workflowParams.get("target_database")
    val targetS3Location = workflowParams.get("target_s3_location")
  }
}

注意事项 #

• AWS Glue托管环境已经预装了所有依赖的AWS SDK和Glue Scala库，无需额外打包依赖即可提交运行
• 通过工作流触发作业时，WORKFLOW_NAME和WORKFLOW_RUN_ID会由Glue自动注入，无需手动在作业参数中配置
• 代码中获取的workflowParams是java.util.Map类型，如需转为Scala原生Map操作，可引入scala.collection.JavaConverters._后调用workflowParams.asScala完成转换

内容的提问来源于stack exchange，提问作者Max08