AWS Glueを用いてパフォーマンス向上やコスト最適化するカラム名ありパーティションのデータに変換するETLコードを作成する

はじめに

Amazon Redshift Spectrum や Amazon Athena は、いかにスキャンデータ量を少なくするかが重要です。その理由は、スキャンデータ量を少なくすることによってコストを削減するだけでなく、パフォーマンスも改善するからです。一般的なRDBのレンジスキャン、Amazon Redshiftのゾーンマップによるスキャンと同様です。スキャンデータ量を少なくするには様々なアプローチがありますが、Amazon Redshift Spectrum や Amazon Athenaにおいてスキャン対象のデータを削減するには、データを事前にパーティションして、条件に基づいてスキャンするパーティションを削減します。今回は、AWS Glueを用いてパフォーマンス向上やコスト最適化するカラム名ありパーティションのデータに変換するETLコードを作成する方法をご紹介します。

パーティションでデータとして保存する理由

逆にパーティションでデータを保存しないということは、単一のフォルダの下に、全てのレコードのファイルを格納した状態です。例えば、単一のフォルダの中に日毎のデータファイルが格納されているとします。この状態で特定の日のデータを条件指定しても常にフォルダの全てのデータをスキャンします。つまり、一般的なRDBを例えるならインデックスを指定していないテーブルに対する sequential scan のような動作となります。

2018-01-01 12:34:56 329872397 mybucket/posdata/posdata-20180101.csv.gz
2018-01-02 12:34:56 397484959 mybucket/posdata/posdata-20180102.csv.gz
2018-01-03 12:34:56 367949672 mybucket/posdata/posdata-20180103.csv.gz
:
:
2018-04-30 12:34:56 357608017 mybucket/posdata/posdata-20180430.csv.gz

パーティションでデータを保存するということは、キー毎にフォルダを分けてデータファイルを格納します。例えば、日毎のフォルダを作成して、その下に日毎のデータファイルが格納します。下記のdata=YYYYMMDDの部分です。この状態で特定の日のデータを条件指定してクエリすると、条件に指定したフォルダの下のデータファイルのみスキャンすることでスキャンの範囲を削減できます。

2018-01-01 12:34:56 329872397 mybucket/posdata/date=20180101/posdata-20180101.csv.gz
2018-01-02 12:34:56 397484959 mybucket/posdata/date=20180102/posdata-20180102.csv.gz
2018-01-03 12:34:56 367949672 mybucket/posdata/date=20180103/posdata-20180103.csv.gz
:
:
2018-04-30 12:34:56 357608017 mybucket/posdata/date=20180430/posdata-20180430.csv.gz

Amazon Redshift Spectrum や Amazon Athenaにおいてスキャン対象のデータを削減するため、パーティションでデータとして保存するのです。

カラム名ありパーティションとは

カラム名ありパーティション（Hive互換）は、フォルダ名がキーバリュー形式になっており、例えばフォルダがdate=YYYYMMDD形式で保存されています。この形式はMSCK REPIRE TABLEを実行するとフォルダを再帰的にスキャンして、パーティションを自動設定できます。データを変換するのであれば、カラム名ありパーティションでデータを保存することで、パーティション管理を自動化できるメリットがあります。

パーティション出力する方法

マネジメントコンソールのJobの設定画面では、パーティション設定の指定ができませんので、生成されたPySparkのETLコードに変更を加えます。下記のWrite Dynamic Frame to sinkの部分です。

方法は、Apache Spark 標準の Dataframeによるパーティション出力する方法と AWS Glue が提供する Dynamicframeによるパーティション出力する方法の２通りあります。RDDと、RDDから派生するDataframeやDynamicframeの関係は以下のとおりです。Dynamicframeは、AWS Glueが提供するETLに特化したフレームワークのようなものです。

Dataframeによるパーティション出力

当初は、Dynamicframeでは、パーティション出力をサポートしていなかった（マニュアルにはなかった）ので、こちらの方法一択でした。Dynamicfram と Dataframe は相互変換できる関数が提供されています。この仕組を利用して最終的にファイル出力する段階でDynamicframeからDataframeに変換します。DataframeのpartitionBy()をメソッドチェーンに追加してファイル出力します。

datasink2 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": "s3://mybucket/posdata"}, format = "csv", transformation_ctx = "datasink2")

生成される上記のコードの代わりに、

df_out = applymapping1.toDF()
partition_keys=['sales_date']
df_out.repartition(*partition_keys).write.partitionBy(partition_keys).mode("append").csv("s3://mybucket/posdata")

のようにDataframeを用いて出力します。

• １行目は、toDF()関数を用いて、Dynamicfram から Dataframe に変換しています。変換してしまえば後は、Apache Spark 標準の DataframeのAPIを利用して出力できます
• ２行目は、パーティションキーを配列に定義します。
• ３行目は、パーティションキーのカーディナリティに従ってパーティション指定（repartion()関数）した後、パーティションキーに従ってパーティション分割（partitionBy()関数）します。

Dynamicframeによるパーティション出力

Dynamicframeの glueContext.write_dynamic_frame.from_options の connection_optionsにpartitionKeysを指定することでDynamicframeでパーティション出力できます。

datasink2 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": "s3://mybucket/posdata"}, format = "csv", transformation_ctx = "datasink2")

生成される上記のコードに、, "partitionKeys": ["sales_date"]を追加するのみです。

datasink2 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": "s3://mybucket/posdata", "partitionKeys": ["sales_date"]}, format = "csv", transformation_ctx = "datasink2")

参考：Managing Partitions for ETL Output in AWS Glue - Writing Partitions

最後に

私がAWS Glueを実務で導入するときにまず調べたのが、本日紹介した「Dataframeによるパーティション出力する方法」でした。現在はDynamicframeが標準でサポートされたので、この機会にご紹介しています。

Dataframeによるパーティション出力する方式のメリットは、Dynamicframeが提供していない機能や、指定がメソッドチェーンで追加できることです。一方、Dynamicframeによるパーティション出力するの方式のメリットは、Dynamicframeが良しなにやってくれている仕組みが利用できることです。Dynamicframeの機能で十分な場合やETLを安定運用したいなら、Dynamicframeを利用することをお薦めします。