はじめに

こんにちは。株式会社ジールの@Shin-Nakamura224です。
先日、Glue Data Quality(プレビュー版)が発表されたので、
概要と簡単な検証を紹介したいと思います。
今回の記事では、Glue Data Qualityを使った簡単な検証を紹介したいと思います。

概要↓
Glue Data Quality(プレビュー版)について①

前提条件

・Glue Data Qualityを扱うために、Glue3.0を使用しております。
・データソースはテストデータ（CSVファイル）を使用しております。

検証内容

以下の2つをテーマとして、検証を実施しました。

① ELTジョブ内でData Qualityの品質検査の実施と
　Data Qualityのルール違反をAWS CloudWatchで通知
② Python処理との実行時間の比較

アーキテクチャ

①の検証では、
AWS S3内のテストデータをソースとしてData Qualityで品質検査をします。
結果が失敗したデータに関しては、失敗のステータスをトリガーとして、
AWS CloudWatchのイベントを起動させて、
AWS SNS経由でメールを送信します。

②の検証では、
Glueジョブを利用して、Pythonスクリプトでデータを分析したときとの
実行時間比較を行います。

検証手順

①ELTジョブ内でData Qualityの品質検査の実施とData Qualityのルール違反をAWS CloudWatchで通知

1.AWS S3バケットにテストデータ（.csv）を準備
2.AWS Glueジョブ作成
3.Evaluate Data Qualityノードを追加して、ルールとその他各種設定を実施
4.AWS SNS topic作成
5.AWS SNS サブスクリプション作成
6.AWS CloudWatch ルール作成
7.ジョブ実行
8.結果確認
9.通知メール確認

②Python処理との実行時間の比較

1.AWS Glueジョブ(Data Quality用)作成
2.Pythonシェルスクリプトファイルの作成
3.AWS Glueジョブ(Python用)作成
4.ジョブ実行
5.実行時間確認・比較

検証

①ELTジョブ内でData Qualityの品質検査の実施とData Qualityのルール違反をAWS CloudWatchで通知

1.AWS S3バケットにテストデータ（.csv）を準備

test_data.csv

バケットにアップロード

2.AWS Glueジョブ作成

今回は、「Visual with source and target」を選択して作成します。

3.Evaluate Data Qualityノードを追加して、ルールとその他各種設定を実施

今回は、成功用と失敗用の2つを用意しました。（青枠：成功　赤枠：失敗）

DQDLルール（今回はカラムの有無を確認する「ColumnExists」関数を使用）
対象カラムは、「name」と存在しない「dept」に設定します。
※左ペインの「DQDL rule builder」から、ルールとデータのカラムを追加することができます。

詳細設定

以下が設定内容となります。

Data quality actions

「Publish results to Amazon CloudWatch」：CloudWatchに品質評価の結果パブリッシュを送信します。
「Fail job when data quality fails」：品質評価の結果が「fails」だった場合に、ジョブを停止します。

Data quality transform output

「Original data」: 品質評価の結果に関係なく、元のデータを出力します。
「Data quality results」：データ品質評価の結果を出力します。
※どちらか片方しか選べません。

Data quality output settings

データ品質評価の結果を出力するAWS S3バケットを任意で指定します。

今回は、
「Publish results to Amazon CloudWatch」
「Data quality results」
「Fail job when data quality fails」にチェックをして、
「Data quality output settings」のパスを指定します。

4.AWS SNS topic作成

タイプは「スタンダード」を選択します。

5.AWS SNS サブスクリプション作成

作成したトピックを開いて、「サブスクリプションの作成」をクリックします。

6.AWS CloudWatch ルール作成

カスタムイベントパターンの構築：下記のイベントパターンを入力します。
ターゲット：「4で作成したトピック」を選択します。

{
  "source": [
    "aws.glue"
  ],
  "detail-type": [
    "Glue Job State Change"
  ],
  "detail": {
    "jobName": [
      "quality-test"
    ],
    "state": [
      "FAILED"
    ]
  }
}

7.ジョブ実行

「Save」> 「Run」をクリックして、ジョブを実行します。

8.結果確認

「Data quality」タブを選択すると、品質評価の結果を確認できます。
「Status」と「Status detail」で内容を確認できます。
今回は、「dept」というカラムがないので、「failed」になっています。

「Data quality output settings」で指定したバケット内に、
品質評価の結果ファイルが出力されていることが確認できます。

S3 selectでファイルの中身を確認します。

9.通知メール確認

メールではジョブ結果の詳細を確認することができます。

②Python処理との実行時間の比較

1.AWS Glueジョブ(Data Quality用)作成

AWS S3バケットにテストデータ(.csv)を準備します。

test_data4（５０万件程のデータ）

新たなジョブを作成して、「Data source - S3 bucket」に準備した「test_data4.csv」のパスを設定します。

DQDLルールを設定して、「Save」をクリックします。
設定したルール：
・ColumnLength "password" = 8
・IsComplete "name"

2.Pythonシェルスクリプトファイルの作成

１で作成したDQDLルールを再現するPythonシェルスクリプトファイルを作成します。

test.py

import pandas as pd

#検証したいファイルのパスを指定
input_file = 's3://nakamura-test-202211/glue-test/input/test_data4.csv'


#検証用の関数を作成
def check_length(df, data_len):
    if df[lambda x: x.str.len() > data_len].count() + df[lambda x: x.str.len() < data_len].count() > 0:
        result = 'False'
    else:
        result = 'True'
    return result

def check_null(df):
    if df.isnull().sum() > 0:
        result = f'False, {df.isnull().sum()} data is null'
    else:
        result = 'True'
    return result
 
test_content = {"tests":[{"Columns":'password',"check_length":8},{"Columns":'name',"check_null":''}]}

df = pd.read_csv(input_file)

for test in test_content['tests']:
    print(test['Columns'])
    test_column = test['Columns']
    try :    
        test_df = df[test_column]
    except:
        print('No such columns.')
        pass

    if list(test)[1] =='check_length' :
        result = check_length(test_df, test.get('check_length'))
        print(result)
    else:
        result = check_null(test_df)
        print(result)

3.AWS Glueジョブ(python shell用)作成

新たにジョブを作成して、２で作成したスクリプトファイルの中身をジョブ内の「Script」に貼り付けて、「Save」をクリックします。

4.各ジョブ実行

３までの準備が終了したら、作成した各ジョブを実行します。

5.実行時間確認・比較

Data qualityのジョブ結果
(Worker Type G.1Xを使用しているため、DPUは最小の２で設定)

Pythonスクリプトのジョブ結果

まとめと所感

今回はGlue Data Qualityを用いて、2つの検証を実施しました。
①ELTジョブ内でData Qualityの品質検査の実施とData Qualityのルール違反をAWS CloudWatchで通知
②Python処理との実行時間の比較

検証を通しての所感として、以下の2つが挙げられます。
・DQDLのルール設定が簡単で、
　他の方法に比べて、簡単にデータの品質検査を行うことができるのがとても便利。
・Python処理に比べると実行時間は長くなるため、パフォーマンスはあまり期待できない。

⇒経験が浅いユーザやもともとDeequを利用していたユーザにおすすめ

今回は2回に渡って AWS Glue Data Quality について紹介いたしました。
CloudWatchとの連携や実行時間の比較についての記事がなかったため、
今回の記事を投稿いたしました。
この記事が少しでも皆様のお役に立てれば幸いです。