Glue Data Qualityで品質検査ルールに不合格となったレコードを特定する

品質検査ルールで不合格のレコードをAmazon Athenaを使って特定します

#AWS

#AWS Glue

#AWS Glue Data Quality

niino

2024.06.24

データアナリティクス事業本部インテグレーション部機械学習チーム・新納（にいの）です。

データのETLを担うサービスであるAWS Glueの機能群の一つ、Glue Data QualityではGUIベースの操作でデータの品質検査が可能です。データレイクやデータウェアハウスへのデータ投入前に、値の長さチェックや特定カラムがNULLでないかどうかなど、あらかじめ定義しておいたルールに合格しているかどうかを確認できます。

今回は、定義したルールに不合格となったレコードを特定する方法を検証してみました。

Glue Data Qualityについておさらい

ユーザーが定義したルールに従ってデータの品質検査をできるAWS Glueの機能のひとつです。AWSが開発したOSSであるDeequを使用しており、ルールの定義にはDQDL（Data Quality Definition Language）を利用します。

詳細は以下をご参照ください。

以下のエントリでは動画で実際の操作をご覧いただけます。

ルールに不合格となったレコードを特定する

シナリオ

今回は以下のような架空のユーザーデータを使って、nameがNULLのものはないか、credit_cardが全て16桁であるかチェックします。このルールに合致しないレコードをAmazon Athenaを使って特定します。

"name","name_hiragana","age","birthday","gender","email","phone_number","zipcode","address","company","credit_card","exp_date"
"鈴木 賢司","すずき けんじ",72,"1952-06-11","M","[email protected]","04-3537-7908","337-1051","埼玉県川口市南鳩ヶ谷2-3-10","","4684847007028235","06/27"
"小宮 真由子","こみや まゆこ",50,"1974-04-07","F","[email protected]","0567-38-4614","453-3750","愛知県春日井市高蔵寺町1-5-13","合同会社ＫＡＮＤＩＣａｐｉｔａｌＭａｎａｇｅｍｅｎｔ","3574517766884178","05/28"
"堀 勇希","ほり ゆうき",55,"1968-11-13","M","[email protected]","0228-18-2254","988-2801","宮城県仙台市泉区松陵1-2-5","株式会社ゼウス","3540484972504721","02/28"
"佐藤 涼","さとう りょう",31,"1992-12-02","M","[email protected]","0823-74-1579","725-9090","広島県呉市本通2-3-13","有限会社ビクトリー","3564069950438371","05/26"
"長浜 裕子","ながはま ゆうこ",24,"2000-05-10","F","[email protected]","0466-71-5467","233-9894","神奈川県横浜市旭区笹野台1-1-17","合同会社エバレッジ","372675929782430","10/25"
"前田 雅也","まえだ まさや",45,"1979-04-05","M","[email protected]","03-1493-8474","118-4100","東京都台東区松が谷2-1-3ユーハウス106","","4378069700369912","01/26"
"片山 隼","かたやま はやと",30,"1993-12-29","M","[email protected]","03-1912-4421","123-8031","東京都江東区亀戸2-1-6","合同会社安清","371168643360445","02/25"
...（中略）...
"蔵 めそ","くら めそ",30,"1993-11-30","M","[email protected]","06-9636-4964","586-5852","大阪府大阪市中央区","","13584595506048971","09/28"

Glue Job作成

Data QualityはGlue Data CatalogとGlue Jobのどちらかから利用可能です。今回はGlue Job（Visual Job）のを利用し、S3に配置した上記のファイルを品質チェックしてその結果をS3バケットに出力します。

SourceにはAmazon S3を指定し、インプットデータのURIを指定します。

Evaluate Data Qualityノードを追加してデータ品質ルールを定義します。 Ruleset editorに以下の通りDQDLを定義します。

Rules = [
    ColumnLength "credit_card" = 16,
    IsComplete "name"
]

Data quality transform outputではデータ品質検査の出力について定義します。

項目	内容
Original data	有効にすると元の入力データを出力する。
Add new columns to indicate data quality errors	Original dataを有効にした場合のみ選択可能。有効にすると以下のカラムを元の入力データに追加して出力する。 dataqualityrulespass：合格したルール dataqualityrulefail：不合格になったルール dataqualityrulesskip：スキップされたルール。ただし、次のルールはデータセットレベルで有効なため、不合格レコードを特定できない。 AggregateMatch・ColumnCount・ColumnExists・ColumnNamesMatchPattern・CustomSql・RowCount・RowCountMatch・StandardDeviation・Mean・ColumnCorrelation dataqualityevaluationresult：ルールの合否（Failed/Passed）
Data quality results	有効にするとルールと合否ステータスを出力する。

上記の設定は全て有効にして、最終的なアウトプットにData Qualityの結果カラムを追加されるように設定しました。

Data quality actionsでは、ルールが不合格だった場合にジョブをどうするか設定可能です。「Continue with job（ジョブを続行する）」、「Fail job after loading target data（ターゲットデータをロードしてからジョブを異常終了する）」のどちらかを選んでください。「Fail job without loading target data（ターゲットデータをロードせずジョブを異常終了する）」を選択するとData Qualityの結果カラムが出力されなくなります。今回はロードしてから異常終了する方を選択しました。

Original dataを有効にするとrowLevelOutcomesノードが、Data quality resultsを有効にするとruleOutcomesノードがそれぞれ自動で追加されます。この2つのノードには手動で設定を追加する必要はありません。

それぞれの結果をS3バケットに保存するため、Data targetとしてAmazon S3ノードを追加します。rowLevelOutcomesの出力結果を使って不合格レコードを特定するため、わかりやすいようにruleOutcomesと出力先のフォルダは分けておきましょう。

設定が完了したらSaveからジョブを保存し、ジョブを実行します。今回は「Fail job after loading target data（ターゲットデータをロードしてからジョブを異常終了する）」を選択しており、ルールに適合しないレコードがあるため、Error Category: UNCLASSIFIED_ERROR; AssertionError: The job failed due to failing DQ rules for nodeというエラーでジョブ自体が異常終了します。

Data qualityタブを見ると、ColumnLength "credit_card" = 16のルールが不合格だったことが確認できます。どのレコードが不合格だったのか特定してみましょう。

rowLevelOutcomesの出力結果を保存したファイルを確認すると、元の入力データにData Qualityの結果カラムが追加されていることが確認できます。このままでは確認しづらいため、Amazon Athenaからクエリできるように設定してみましょう。

Amazon Athenaで不合格レコードを特定

今回はGlue crawlerを使って、rowLevelOutcomesの出力先からテーブルを作成しました。テーブル定義は以下の通り。

CREATE EXTERNAL TABLE `user_jpoutput`(
  `name` string, 
  `name_hiragana` string, 
  `age` string, 
  `birthday` string, 
  `gender` string, 
  `email` string, 
  `phone_number` string, 
  `zipcode` string, 
  `address` string, 
  `company` string, 
  `credit_card` string, 
  `exp_date` string, 
  `dataqualityrulespass` array<string>, 
  `dataqualityrulesfail` array<string>, 
  `dataqualityrulesskip` array<string>, 
  `dataqualityevaluationresult` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://<rowLevelOutcomesの出力先S3バケット>/'
TBLPROPERTIES (
  'CrawlerSchemaDeserializerVersion'='1.0', 
  'CrawlerSchemaSerializerVersion'='1.0', 
  'UPDATED_BY_CRAWLER'='nino-dq-test', 
  'averageRecordSize'='197', 
  'classification'='parquet', 
  'compressionType'='none', 
  'objectCount'='2', 
  'recordCount'='2002', 
  'sizeKey'='221694', 
  'typeOfData'='file')

作成されたテーブルを見ると、dataqualityrulespass、dataqualityrulefail、dataqualityrulesskip、dataqualityevaluationresultの4つのData Qualityの結果カラムが追加されていることが確認できます。

WHERE "dataqualityevaluationresult" = 'Failed'でクエリすればルールに不合格となったレコードを特定することができました。