如何获取AWS Glue中CustomSql数据质量检查的失败行

问题 #

AWS Glue已支持检测CustomSql数据质量规则校验失败的行，但执行现有代码仅能看到失败占比，无法查看具体的失败行数据。

当前代码与执行结果 #

执行代码 #

from pyspark.sql import SparkSession
from awsglue.context import GlueContext
from awsglue.dynamicframe import DynamicFrame
from awsgluedq.transforms import EvaluateDataQuality
from pyspark.sql import Row


spark = SparkSession.builder.getOrCreate() 
spark.conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')

glue_ctx = GlueContext(spark.sparkContext)

df = spark.createDataFrame([
    Row(campagne='dzdzd', num=5),
    Row(campagne='["campagne1", "campagne2"]', num=5),
    Row(campagne='["campagne1", "campagne2"]', num=1)
])
dynamic_df = DynamicFrame.fromDF(df, glue_ctx)

# 执行数据质量校验
dq_result = EvaluateDataQuality.apply(
        frame=dynamic_df,
        ruleset="""
    Rules = [
        CustomSql "select campagne from primary where num = 5"
    ]
    """,
        publishing_options={}
)

# 仅查看规则结果
dq_result.toDF().show(truncate=False)

现有执行结果 #

+------------------------------------------------------+-------+---------------------------------------------------+------------------------------------------------------+------------------------------------------------------+
|Rule                                                  |Outcome|FailureReason                                      |EvaluatedMetrics                                      |EvaluatedRule                                         |
+------------------------------------------------------+-------+---------------------------------------------------+------------------------------------------------------+------------------------------------------------------+
|CustomSql "select campagne from primary where num = 5"|Failed |Custom SQL response failed to satisfy the threshold|{Dataset.*.CustomSQL.Compliance -> 0.6666666666666666}|CustomSql "select campagne from primary where num = 5"|
+------------------------------------------------------+-------+---------------------------------------------------+------------------------------------------------------+------------------------------------------------------+

解决方案 #

要获取具体的失败行，需要在EvaluateDataQuality.apply中添加additional_options参数启用失败记录收集，然后从返回的结果集合中提取失败记录帧：

修改后的代码 #

from pyspark.sql import SparkSession
from awsglue.context import GlueContext
from awsglue.dynamicframe import DynamicFrame
from awsgluedq.transforms import EvaluateDataQuality
from pyspark.sql import Row


spark = SparkSession.builder.getOrCreate() 
spark.conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')

glue_ctx = GlueContext(spark.sparkContext)

df = spark.createDataFrame([
    Row(campagne='dzdzd', num=5),
    Row(campagne='["campagne1", "campagne2"]', num=5),
    Row(campagne='["campagne1", "campagne2"]', num=1)
])
dynamic_df = DynamicFrame.fromDF(df, glue_ctx)

# 启用失败记录收集，执行数据质量校验
dq_result_collection = EvaluateDataQuality.apply(
        frame=dynamic_df,
        ruleset="""
    Rules = [
        CustomSql "select campagne from primary where num = 5"
    ]
    """,
        publishing_options={},
        additional_options={"enableDataQualityResultRecords": True}
)

# 提取规则结果和失败记录
rule_outcomes = dq_result_collection.select("ruleOutcomes").toDF()
failing_records = dq_result_collection.select("failingRecords").toDF()

# 查看规则结果
print("=== 数据质量规则结果 ===")
rule_outcomes.show(truncate=False)

# 查看具体失败行
print("\n=== 校验失败的行记录 ===")
failing_records.show(truncate=False)

预期执行结果 #

=== 数据质量规则结果 ===
+------------------------------------------------------+-------+---------------------------------------------------+------------------------------------------------------+------------------------------------------------------+
|Rule                                                  |Outcome|FailureReason                                      |EvaluatedMetrics                                      |EvaluatedRule                                         |
+------------------------------------------------------+-------+---------------------------------------------------+------------------------------------------------------+------------------------------------------------------+
|CustomSql "select campagne from primary where num = 5"|Failed |Custom SQL response failed to satisfy the threshold|{Dataset.*.CustomSQL.Compliance -> 0.6666666666666666}|CustomSql "select campagne from primary where num = 5"|
+------------------------------------------------------+-------+---------------------------------------------------+------------------------------------------------------+------------------------------------------------------+

=== 校验失败的行记录 ===
+------------------------------+---+
|campagne                      |num|
+------------------------------+---+
|["campagne1", "campagne2"]    |1  |
+------------------------------+---+

关键说明 #

• additional_options={"enableDataQualityResultRecords": True}：开启该参数后，AWS Glue会收集所有未通过CustomSql规则的行记录
• 返回的dq_result_collection是一个DynamicFrameCollection，包含两个核心帧：
  • ruleOutcomes：原有的数据质量规则校验结果
  • failingRecords：所有未通过校验的原始行数据

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