将数据质量结果写入 Data Catalog 表

您可以将 AWS Glue 数据质量评估运行配置为自动将结果写入 AWS Glue Data Catalog 中的 Apache Iceberg 表。启用结果输出后，您可以直接使用相关工具查询数据质量结果，使用可视化工具构建控制面板，并在整个账户中集中维护数据质量结果的历史记录。

您可以将以下类型的数据质量结果写入 Data Catalog 表：

先决条件

要将数据质量结果写入 Data Catalog 表，您用于评估运行的 IAM 角色必须具有以下权限：

评估运行使用您指定的 IAM 角色写入结果表。这与有权访问源数据表的角色相同。

配置结果输出

您可以使用 StartDataQualityRulesetEvaluationRun API 的 --additional-run-options 参数或 AWS Glue ETL 作业中的 additional_options 参数来配置数据质量结果输出。默认情况下，AWS Glue 数据质量自动监测功能数据质量自动监测功能不会将结果写入 Data Catalog 表。您必须明确启用要写入的每种结果类型。

每种结果类型都有其自己的配置块，并具有共享 CatalogTableConfig 结构。如果您不提供 CatalogTableConfig，AWS Glue 数据质量自动监测功能会自动派生默认值，包括表名称和 Amazon S3 路径。

CatalogTableConfig 结构包含以下字段：

示例：配置规则结果和分析结果

aws glue start-data-quality-ruleset-evaluation-run \
  --data-source '{
    "GlueTable": {
      "DatabaseName": "my_database",
      "TableName": "my_table"
    }
  }' \
  --role "arn:aws:iam::123456789012:role/GlueServiceRole" \
  --ruleset-names '["my_ruleset"]' \
  --additional-run-options '{
    "DataQualityRuleResults": {
      "WriteDataQualityRuleResultsEnabled": true,
      "CatalogTableConfig": {
        "DatabaseName": "quality_results",
        "TableName": "rule_results"
      }
    },
    "ProfilingResults": {
      "WriteProfilingResultsEnabled": true,
      "CatalogTableConfig": {
        "DatabaseName": "quality_results",
        "TableName": "profiles"
      }
    }
  }'

示例：配置行级结果

对于行级结果，您还可以指定要包含的记录类型以及要写入的最大行数。

aws glue start-data-quality-ruleset-evaluation-run \
  --data-source '{
    "GlueTable": {
      "DatabaseName": "my_database",
      "TableName": "my_table"
    }
  }' \
  --role "arn:aws:iam::123456789012:role/GlueServiceRole" \
  --ruleset-names '["my_ruleset"]' \
  --additional-run-options '{
    "RowLevelResults": {
      "MaxRowsToWrite": 5000,
      "ResultType": "FAILED_ONLY",
      "CatalogTableConfig": {
        "DatabaseName": "quality_results",
        "TableName": "row_level_results"
      }
    }
  }'

ResultType 参数接受以下值：

示例：在 AWS Glue ETL 作业中进行配置

在 AWS Glue ETL 作业中，您可以使用带点符号键的 additional_options 参数配置结果输出：

result = EvaluateDataQuality.process_rows(
    frame=dynamic_frame,
    ruleset=ruleset,
    publishing_options={
        "dataQualityEvaluationContext": "my_context",
        "enableDataQualityResultsPublishing": True
    },
    additional_options={
        "observations.scope": "ALL",
        "dataQualityResultsPublishing.strategy": "BEST_EFFORT",
        "dataQualityResultsPublishing.resultsFormat.profilingResults.writeProfilingResultsEnabled": "true",
        "dataQualityResultsPublishing.resultsFormat.profilingResults.catalogTableConfig.databaseName": "my_db",
        "dataQualityResultsPublishing.resultsFormat.profilingResults.catalogTableConfig.tableName": "profiling_results",
        "dataQualityResultsPublishing.resultsFormat.profilingResults.catalogTableConfig.s3Location": "s3://amzn-s3-demo-bucket/profiling/",
        "dataQualityResultsPublishing.resultsFormat.profilingResults.catalogTableConfig.catalogId": "123456789012"
    }
)

示例：配置观测结果

您可以像配置其他结果类型一样配置观测结果。观测结果需要启用异常检测功能（ObservationScope: ALL）：

aws glue start-data-quality-ruleset-evaluation-run \
  --data-source '{
    "GlueTable": {
      "DatabaseName": "my_database",
      "TableName": "my_table"
    }
  }' \
  --role "arn:aws:iam::123456789012:role/GlueServiceRole" \
  --ruleset-names '["my_ruleset"]' \
  --additional-run-options '{
    "ObservationScope": "ALL",
    "ObservationResults": {
      "WriteObservationResultsEnabled": true,
      "CatalogTableConfig": {
        "DatabaseName": "quality_results",
        "TableName": "observation_results"
      }
    }
  }'

表架构

AWS Glue 数据质量自动监测功能将每种结果类型写入单独的 Iceberg 表。规则结果、分析结果（包括单独的分布结果表）和观测结果表按 catalog_id、database_name、table_name 和 day(stored_on) 进行分区，以实现高效查询。您可以直接对 stored_on 进行筛选以执行基于时间的查询，Iceberg 会自动处理分区修剪。

规则结果表

规则结果表存储在数据质量运行期间评估的每条规则的通过或失败结果。

数据剖析结果表

下表描述数据剖析结果表中的各列。此表存储分析器和规则（例如 Mean、StandardDeviation 和 Completeness）收集的标量统计数据。AWSGlue 数据质量自动监测功能将分布统计数据存储在单独的分布结果表中。

分布结果表

下表介绍了分布结果表中的各列。分布结果与标量分析统计数据分开存储，每个分箱或类别对应一行。您可以在 ProfilingResults.DistributionResults 块中配置此表。

行级结果表

下表描述行级结果表中的各列。您可以使用此表来标识不符合数据质量规则的特定记录。

观测结果表

观测结果表存储每次评估运行中每个统计数据的异常检测预测结果。该表包含所有预测结果：异常值、正常值和跳过的预测。这使您可以呈现带有预测区间的连续趋势图。

查询结果

数据质量评估完成后，您可以直接查询结果表。以下示例演示常见的查询模式。

示例：查找特定运行的失败规则

SELECT rule_name, rule_description, evaluation_message, evaluated_metrics
FROM quality_results.rule_results
WHERE ruleset_evaluation_run_id = 'dqr-12345678'
  AND rule_result = 'FAIL'
ORDER BY rule_name;

示例：查看一段时间内的分析统计数据

SELECT stored_on, statistics_value
FROM quality_results.profiles
WHERE database_name = 'my_database'
  AND table_name = 'my_table'
  AND statistic_name = 'Mean'
  AND columns_referenced = ARRAY['salary']
ORDER BY stored_on;

示例：识别未通过特定规则的行

SELECT *
FROM quality_results.row_level_results
WHERE data_quality_evaluation_result = 'Failed'
  AND contains(data_quality_rules_fail, 'IsComplete "email"');

示例：查看数字直方图

SELECT bin_index, bin_lower_bound, bin_upper_bound, bin_count
FROM quality_results.distributions
WHERE column_name = 'salary'
  AND ruleset_evaluation_run_id = 'dqrun-abc123'
ORDER BY bin_index;

示例：查看类别值分布

SELECT bin_label, bin_count
FROM quality_results.distributions
WHERE column_name = 'department'
  AND ruleset_evaluation_run_id = 'dqrun-abc123'
ORDER BY bin_count DESC;

示例：跟踪一段时间内的类别频率

SELECT started_on, bin_count
FROM quality_results.distributions
WHERE column_name = 'status' AND bin_label = 'active'
ORDER BY started_on;

示例：使用预测区间查看异常检测趋势

SELECT o.recorded_on, 
       p.statistics_value AS actual_value,
       o.expected_value, 
       o.lower_bound, 
       o.upper_bound,
       o.prediction_outcome
FROM quality_results.profiles p
JOIN quality_results.observation_results o
  ON p.statistic_id = o.statistic_id
  AND p.ruleset_evaluation_run_id = o.ruleset_evaluation_run_id
WHERE p.database_name = 'my_database'
  AND p.table_name = 'my_table'
  AND p.statistic_name = 'RowCount'
  AND p.stored_on >= DATE '2025-03-01'
ORDER BY p.stored_on;

示例：注释后查询最新观测状态

由于观测结果表使用仅追加模型，排除注释会添加新行。使用重复数据删除查询获取每次观测的最新状态：

SELECT statistic_id, statistic_name, prediction_outcome, 
       expected_value, lower_bound, upper_bound, training_input, stored_on
FROM (
  SELECT *, ROW_NUMBER() OVER (
    PARTITION BY statistic_id, ruleset_evaluation_run_id 
    ORDER BY stored_on DESC
  ) AS rn
  FROM quality_results.observation_results
  WHERE database_name = 'my_database'
    AND table_name = 'my_table'
)
WHERE rn = 1
ORDER BY stored_on;

注意事项

将数据质量结果写入 Data Catalog 表时，请牢记以下注意事项：