如何在AWS Glue Scala任务中减少AWS Deequ校验代码重复

核心设计逻辑 #

不需要写多份几乎一模一样的任务脚本，只要把所有不变的流程（Glue上下文初始化、Deequ规则校验、结果格式转换）封装成独立的公共方法，把每个数据集不一样的参数（输入S3路径、输出S3路径、数据集名称标识）抽出来做成可配置的清单，从根源消掉重复代码。后续数据集数量涨多少，只要往配置清单里加一行就行，核心校验逻辑完全不用碰。

第一步：抽离无重复的公共校验方法 #

把所有数据集通用的校验逻辑、读数逻辑全部封装为独立的可调用方法，不绑定任何固定的数据集路径，所有可变参数以入参形式传入：

import com.amazon.deequ.VerificationSuite
import com.amazon.deequ.checks.{Check, CheckLevel}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.lit

/**
 * 通用Deequ校验逻辑，所有数据集共用同一套规则
 * @param spark 复用已初始化的Spark会话
 * @param inputS3Path 目标数据集的S3输入路径
 * @param datasetName 数据集唯一标识，用于结果溯源
 * @return 结构化的校验结果DataFrame
 */
def runCommonDqCheck(
  spark: SparkSession,
  inputS3Path: String,
  datasetName: String
): DataFrame = {
  // 固定读数逻辑：因所有数据集结构、表头完全一致，无需重复定义schema
  val inputDF = spark.read
    .option("header", "true")
    .csv(inputS3Path)

  // 固定校验规则：所有数据集执行同一套Deequ校验
  val checkResult = VerificationSuite()
    .onData(inputDF)
    .addCheck(
      Check(CheckLevel.Error, "通用校验规则集")
        .hasSize(_ >= 1000, Some("数据集行数不低于1000的校验"))
        .isNotNull("user_id", Some("user_id非空校验"))
        .isUniqueValue("order_id", Some("order_id唯一性校验"))
    )
    .run()

  // 固定结果转换逻辑：将Deequ原生输出转为结构化DataFrame，追加数据集标识列
  com.amazon.deequ.AnalyzerContext
    .successMetricsAsDataFrame(spark, checkResult)
    .withColumn("dataset_name", lit(datasetName))
}

第二步：维护轻量数据集配置清单 #

用Scala基础的样例类存储每个数据集的可变参数，不用引入复杂的配置管理框架，新手可直接上手：

// 轻量配置实体，存储每个数据集的可变参数
case class DatasetParam(
  datasetName: String,
  inputPath: String,
  outputPath: String
)

// 数据集配置列表，后续新增数据集直接在列表中追加条目即可，核心逻辑零改动
val targetDatasetList = List(
  DatasetParam(
    datasetName = "user_profile",
    inputPath = "s3://your-bucket/raw/user_profile/",
    outputPath = "s3://your-bucket/dq-result/user_profile/"
  ),
  DatasetParam(
    datasetName = "order_info",
    inputPath = "s3://your-bucket/raw/order_info/",
    outputPath = "s3://your-bucket/dq-result/order_info/"
  ),
  DatasetParam(
    datasetName = "log_behavior",
    inputPath = "s3://your-bucket/raw/log_behavior/",
    outputPath = "s3://your-bucket/dq-result/log_behavior/"
  )
)

第三步：主流程批量遍历执行 #

Glue任务启动后仅初始化一次Spark上下文，循环遍历所有配置的数据集，逐个调用公共校验方法，并将结果写入每个数据集对应的独立输出路径：

// 初始化Spark/Glue上下文，全流程仅执行一次
val spark = SparkSession.builder()
  .config(sc.getConf)
  .getOrCreate()

// 遍历所有目标数据集执行校验
targetDatasetList.foreach { dataset =>
  // 调用公共校验逻辑
  val dqResult = runCommonDqCheck(
    spark = spark,
    inputS3Path = dataset.inputPath,
    datasetName = dataset.datasetName
  )
  // 写入对应独立输出路径
  dqResult.write
    .mode("overwrite")
    .option("header", "true")
    .csv(dataset.outputPath)
}

可扩展性优化提示 #

• 后续新增校验规则时，仅需修改runCommonDqCheck方法内的规则集，所有关联的数据集自动继承新规则，无需逐个修改重复代码
• 可以将硬编码的配置列表迁移到Glue Job参数、或S3存储的简单JSON文件中，后续新增数据集时仅需修改配置内容，无需重新打包、部署Scala代码
• 因为所有数据集结构一致，读数逻辑完全复用，不会出现不同任务里重复写schema定义的冗余问题

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