CSVを分割してAmazon S3に保存して、 Amazon Athenaのスキャンコストを下げる(Spark利用)

環境

• Amazon Athena
• Spark 2.4.4
• Apache Zeppelin 0.9.0-preview1
• macOS 10.15 において、 Dockerに環境を構築

Amazon Athena の料金

Amazon Athena は、Amazon S3 から直接データをクエリ処理します。Athena によるデータのクエリ処理に対する追加のストレージ料金は発生しません。ストレージ、リクエスト、データ転送に対して S3 の標準料金が発生します。
Athena で AWS Glue データカタログを使用する場合は、標準の AWS Glue データカタログのレートで課金されます。

引用元

• データを分割して、Athenaがスキャンする対象を明確にすることでコストを下げることができます。

具体例

• 発売年と製品名のデータがあるとします。
• ユーザーは、「2019年に発売した製品の一覧がみたい」などという使い方をするとします。
• データを分割せずに保存した場合は、全てスキャンして、2019年の製品を探すことになります。

+----+---------------------------+
|year|product                    |
+----+---------------------------+
|2018|iPad Pro 12.9 (3rd gen)    |
|2019|MacBook Pro (4th gen)      |
|2020|Magic Keyboard for iPad Pro|
|2020|iPhone SE (2nd gen)        |
+----+---------------------------+

スキャン量を減らすために、年ごとにデータを分割する

• どのツール(独自スクリプトなど)を使ってもよいと思いますが、Sparkを使うと簡単にAthenaと連携できる形式でデータを分割ができます。
• Sparkを使って、年ごとにデータを分割して保存する方法を示します。
• Sparkを使わなくても、分割する形式のみ理解すれば同じことが実現できます。重要なのは、データを分割する部分です。

SparkのScalaのコード

val seq = Seq(
    (2018, "iPad Pro 12.9 (3rd gen)"),
    (2019, "MacBook Pro (4th gen)"),
    (2020, "Magic Keyboard for iPad Pro"),
    (2020, "iPhone SE (2nd gen)")
)
var products = sc.parallelize(seq, 1).toDF("year", "product")
products.write.partitionBy("year").csv("/data/output/products/")

出力したCSVファイルのフォルダ構成

• 年ごとに別のフォルダにCSVが保存されます。
• year=YYYY という形式のフォルダ名になります。
• 年ごとに別のファイル（場所、フォルダ）に保存したという事実をAmazon Athenaに教えてあげれば、スキャン量を減らすことができます。

tree /data/output/apple_products/

/data/output/products/
│
│
├── year=2018
│   └── part-00000-00000000-dd54-40cf-0000000000000000.c000.csv
├── year=2019
│   └── part-00000-00000000-dd54-40cf-0000000000000000.c000.csv
└── year=2020
    └── part-00000-00000000-dd54-40cf-0000000000000000.c000.csv

出力したCSVファイルの中身

• CSVファイルの中身は、yearカラムが存在しません。
• year=YYYYというフォルダ名が重要な情報になります。
• 別のフォルダに移動させる、リネームすると、情報が失われます。
• Amazon Athenaと連携するのに都合がよい形式です。
• 別の要件で、yearカラムを残す必要がある場合は、yearカラムを別名で複製して(year_for_partなど)、パーテーションのキーにして保存すればよいと思いますが、yearカラムを条件指定して検索するとフルスキャンになります。

find . -name "*.csv" | xargs -I% cat %

Magic Keyboard for iPad Pro
iPhone SE (2nd gen)
iPad Pro 12.9 (3rd gen)
MacBook Pro (4th gen)

補足：「partitionBy()メソッド」と「RDDのパーテーション」

• 説明すると逆に混乱を招くかもしれないので、よくわからない方は飛ばしてください。
• partitionBy("year")は、yearごとにデータを分割して、別名のフォルダで保存するために実行しました。
• partitionBy()メソッドの「パーテーション」と「RDDのパーテーション」は、別なので注意してください。
• 以下に示すように、RDD上は、同一のパーテーションです。

+----+---------------------------+------------+
|year|product                    |partition id|
+----+---------------------------+------------+
|2018|iPad Pro 12.9 (3rd gen)    |0           |
|2019|MacBook Pro (4th gen)      |0           |
|2020|Magic Keyboard for iPad Pro|0           |
|2020|iPhone SE (2nd gen)        |0           |
+----+---------------------------+------------+

年ごとに分割したCSVファイルをS3に保存する

• S3に保存します。

AWS Glue Data Catalog の table を作成

• 以下のDDLをAthenaのQuery editorなどで実行します。
• CREATE EXTERNAL TABLE時に、PARTITIONED BY を設定します
• LOCATIONでS3のパスは、通常通り、ルートを指定します。
• 実行してテーブルの表示を確認すると、パーテーションを設定したことが、わかります。

CREATE EXTERNAL TABLE IF NOT EXISTS products (
  `product` string
) PARTITIONED BY (
  year bigint
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
  'serialization.format' = ',',
  'field.delim' = ','
) LOCATION 's3://s3_bucket_name/products/'
TBLPROPERTIES ('has_encrypted_data'='false');

分割したデータの場所を設定するコマンドを実行

• MSCK REPAIR TABLEというコマンドを実行します
• 実行すると、column=value形式のフォルダを探して、パーテーションとしてメタスストアに追加してくれます。
• MSCKは、Hive's metastore consistency checkらしいです。メタストアチェックツールで、テーブルをリペアすると覚えました！
• （何回も打つことになるので、自然に暗記してしまうと思いますが）

参考元 stackoverflow.com

MSCK REPAIR TABLE products

• 以下のような実行結果が表示されます。
• 存在しないパーテーションを自動で修復(Repair)してくれます。
• パーテーションが増えたら、再びこのコマンドを実行する必要があります。

Partitions not in metastore: products:year=2018 products:year=2019 products:year=2020
Repair: Added partition to metastore products:year=2018
Repair: Added partition to metastore products:year=2019
Repair: Added partition to metastore products:year=2020

全件取得

• 全データの容量を把握するために、全件取得してみます。
• 全件スキャンすると、0.09KB　だとわかりました。

SELECT * FROM products;

(Run time: 3.47 seconds, Data scanned: 0.09 KB)

パーテーションを指定したカラムで条件指定

• where year = 2019 を付与して実行してみます。
• スキャン量が、0.02KBになりました
• この例では、実行時間は増えましたが、データが巨大になれば、減る傾向にあると思っています。

SELECT * FROM products where year = 2019;

 (Run time: 3.97 seconds, Data scanned: 0.02 KB)

パーテーションの確認コマンド

• 以下のコマンドでパーテーションの設定を確認できます。
• S3のロケーションの設定を確認したいのですが、現状、確認方法がわかっていません。
• (MSCK REPAIR TABLEでロケーションを指定した場合は、間違うことは少ないので、あまりロケーションを意識しなくてもよいと思います。）

 SHOW PARTITIONS products;

year=2019
year=2020
year=2018

手動でパーテーションとデータの場所を紐付ける

• column=value形式のフォルダ名ではない場合は、「カラム名と値」と「S3のロケーション」の紐付けを手動で行う必要があります。

ALTER TABLE table ADD
PARTITION (year = 2019) LOCATION 's3://s3_bucket_name/2019/'

参考サイト(docs.aws.amazon.com)

まとめ

• 必要に応じて、データを分割して、スキャン対象を減らすと、時間的、費用的なコストを下がると思います。
• CSV形式の場合、各レコードに年数を持つよりも、フォルダ名として１つ保存する方がディスク消費量、スキャン量、転送量が減ります。
• あまりにも細かく分割して、大部分を取得するようなSQLを実行すると、たくさんのファイルを読むことになると思います。