Amazon Athena のパフォーマンスチューニング Tips トップ 10

2020/10/13 に、原文の更新に合わせて最新のバージョンにアップデートしました

Amazon Athena は、S3 に保存されたデータに対して標準 SQL で簡単に分析を行える、インタラクティブクエリサービスです。Athena はサーバーレスのためインフラ管理の必要がなく、また実行したクエリのぶんだけ料金を支払うかたちになります。Athena は簡単に使えます。Amazon S3 上のデータに対してスキーマを定義し、標準 SQL でクエリを投げるだけです。

このブログポストでは、クエリパフォーマンスを改善するための 10 個の Tips をご紹介します。Tips には、Amazon S3 に置かれたデータに関するものと、クエリチューニングに関するものがあります。Amazon Athena は Presto を実行エンジンとして使用しているため、ここでご紹介する Tips のうちのいくつかは、Amazon EMR 上で Presto を動かす際にも当てはまります。

このポストは、読者の方が Parquet, ORC, Text files, Avro, CSV, TSV, and JSON といった、さまざまなファイルフォーマットについての知識を持っていることを前提としています。

ストレージのベストプラクティス

このセクションでは Athena を最大限に活用するために、どのようなデータ構造にするべきかについて議論します。ここで議論する内容は、Amazon EMR 上の Spark, Presto, Hive で Amazon S3 のデータを処理する場合にも、同様に当てはまります。

1. データをパーティションに分ける

パーティショニングとは、テーブルをいくつかに分割し、日付や国、地域といったカラムの値単位でまとめることをさします。パーティションは仮想カラムとして動作します。パーティションをテーブルの作成時に定義することで、クエリでスキャンするデータ量を減らすことができ、その結果パフォーマンスが向上します。パーティションに基づいてフィルタを指定することで、クエリで読み込むデータ量を制限することができます。より詳細については、Partitioning Data を参照してください。

Athena では Hive のパーティショニングをサポートしており、以下のいずれかの記法を使用します。

1. カラム名のあとに = 記号をつけ、そのあとに値を記述する

   s3://yourBucket/pathToTable/<PARTITION_COLUMN_NAME>=<VALUE>/<PARTITION_COLUMN_NAME>=<VALUE>/

   データセットがこの形でパーティションわけされている場合には、テーブルにパーティションを一括で認識させるために、MSCK REPAIR TABLE を実行します

2. もしデータの “Path” が上記のフォーマットでない場合には、個々のパーティションに対して、ALTER TABLE ADD PARTITION コマンドを実行することで、パーティションを認識させることができます

   s3://yourBucket/pathToTable/YYYY/MM/DD/

   Alter Table <tablename> add Partition (PARTITION_COLUMN_NAME = <VALUE>, PARTITION_COLUMN2_NAME = <VALUE>) LOCATION ‘s3://yourBucket/pathToTable/YYYY/MM/DD/’;

   注意: このやり方を用いることで、S3 上のどの場所にあるオブジェクトに対してでも、パーティションを認識させることができます

以下の例は、S3 バケットに置かれたフライトテーブルが、year でパーティションわけされているものになります。

$ aws s3 ls s3://athena-examples/flight/parquet/
PRE year=1987/
PRE year=1988/
PRE year=1989/
PRE year=1990/
PRE year=1991/
PRE year=1992/
PRE year=1993/

year カラムに対して ‘WHERE’ 句を用いることで、読み込むデータ量を制限できます。

SELECT dest, origin FROM flights WHERE year = 1991

パーティションキーには、複数のカラムを指定することができます。同様に、そのカラムが特定の値のものだけをスキャンすることができます。

s3://athena-examples/flight/parquet/year=1991/month=1/day=1/
s3://athena-examples/flight/parquet/year=1991/month=1/day=2/

パーティション対象のカラムを決める際には、以下の点を考慮してください:

• フィルタとして使われるカラムは、パーティションの有力候補になります
• パーティショニング自体にコストがかかります。テーブル内のパーティション数が増えるにつれ、パーティションのメタデータを取得して処理するためのオーバーヘッドが大きくなり、かつ 1 パーティションあたりのデータサイズは小さくなります。過剰に細かくパーティショニングすると、パフォーマンス上の利点が失われます
• データが特定パーティションに偏っており、かつ多くのクエリがその値を使う場合には、同様にパフォーマンス上の利点が失われます

例:

以下のテーブルでは、パーティション分けされたテーブルと、そうでないテーブルのクエリ実行時間を比較しています。両テーブルには、無圧縮のテキストフォーマットデータが 74GB 含まれています。パーティション分けされたテーブルは、l_shipdate カラムによって 2526 個のパーティションに分けられています。

ただし以下に示すように、パーティショニングにはペナルティもあります。過剰にパーティション分けしないように気をつけてください。

2. バケッティングでデータを分割する

パーティションに分ける以外で、データを分割する方法として、バケッティングがあります（訳注: ここでいうバケッティングは，単語としては似ていますが，S3 のバケットとは全く関係がない点に注意してください。原文では bucket と書かれていますが、この bucket は S3 上のファイルのことを指しているため、ここではファイルと訳しています）。バケッティングでは、ひとつ以上のカラムを指定して、それらのカラムの値に基づいてデータを複数のファイルに分けます。つまり、指定したカラムの値が同じレコードは、すべて同じファイルに書き出されます。この処理によって、カラムに特定の値を含むレコードのみをクエリ対象としたときに、読み込むデータ量を大幅に削減することができます。

バケッティングで指定するカラムには、カーディナリティの高いもの（つまり、ユニークな値をとても多く含むカラムということです）、かつクエリを実行する際によくフィルタ対象として使われるものを選ぶようにしてください。バケッティング対象カラムの良い例としては、ユーザー ID のようなプライマリキーが挙げられます。

Athena の CREATE TABLE 句の中で，CLUSTERED BY (<バケッティング対象カラム名>) INTO <結果を分割したいファイル数> と指定することで、バケッティングを実行することができます。分割された各ファイルのサイズが、最適な大きさになるようにファイル数を指定してください。詳細は、4. ファイルサイズを最適化する、を参照してください。

Athena でバケッティングを活用する際には、Apache Hive 形式の CREATE TABLE を実施する必要があります。Apache Spark のバケッティングフォーマットは Athena ではサポートされていない点にご注意ください。バケッティングによるテーブルを作成の詳細は、Apache Hive ドキュメント内の LanguageManual DDL BucketedTables を参照してください。また Athena ではINSERT INTO がサポートされていない点にも気をつけてください。

以下の表は、c_custkey を指定したバケッティングで、32 ファイルにデータを分割したカスタマーテーブルに対して、クエリを実行したときのパフォーマンスを示しています。カスタマーテーブルのサイズは全部合計して 2.29GB です。

3. ファイルを圧縮・分割する

各ファイルが適切なサイズ（詳しくは次のセクションを参照）であるか、ファイルが分割可能であれば、データの圧縮によってクエリの実行速度は著しく向上します。データサイズが小さいほど、S3 から Athena へのネットワークトラフィックが軽減されます。

分割可能なファイルの場合には並列実行性を増すために、Athena の実行エンジンがファイルを分割して、複数のリーダーで処理します。単一の分割不可能なファイルを扱う場合には、単一リーダーのみがファイルを読み込むことができ、そのほかのリーダーは待機状態のままです。すべての圧縮アルゴリズムが分割可能なわけではありません。一般的な圧縮フォーマットとその属性について、以下のテーブルにまとめました。

一般的に、アルゴリズムの圧縮率が高くなるほど、圧縮および解凍に必要な CPU リソースが増えます。

Athena の場合は、デフォルトでデータ圧縮が行われ、かつ分割可能な Apache Parquet や Apache ORC といったファイルフォーマットの利用をおすすめします（訳注: ここでは、ファイルフォーマットと圧縮フォーマットが混ざっている点に注意してください。Parquet や ORC はファイルフォーマットであり、圧縮フォーマットではありません。ですが、Parquet や ORC はデフォルトのエンコーディング法の中に、辞書エンコーディングという簡単な圧縮処理が含まれています。ここの記述は、そのことを表しています。また Tips の 4 番目で述べているように、Parquet/ORC に対して、さらに Snappy のような圧縮アルゴリズムを指定することが可能です）。もしそれらを利用できない場合には、適切なサイズに分割した BZip2 や Gzip を試してください。

from awsglue.job import Job
from awsglue.transforms import *
from awsglue.context import GlueContext
from awsglue.utils import getResolvedOptions

from pyspark.context import SparkContext

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

## Read TABLE_NAME from DB_NAME out of the AWS Glue Data Catalog
dataset = glueContext.create_dynamic_frame.from_catalog(database = DB_NAME, table_name = TABLE_NAME, transformation_ctx = "dataset")

## Write data in JSON format to S3, compressed with GZip
outputdf = glueContext.write_dynamic_frame.from_options( \
	frame = dataset,
	connection_type = "s3",
	connection_options = {"path":"s3://bucket/prefix/"},
	format = "json",
	compression = "gzip",
	transformation_ctx = "outputdf")

job.commit()

4. ファイルサイズを最適化する

データ読み込みが並列で行われ、データブロックがシーケンシャルに読み込まれる場合に、クエリが効率的に実行されます。分割可能なファイルフォーマットであるようにしておくことで、ファイルの大きさに関わらず並列処理が行われます。

ただしファイルサイズが非常に小さい場合、特に 128MB 未満の場合には、実行エンジンは S３ファイルのオープン、ディレクトリのリスト表示、オブジェクトメタデータの取得、データ転送のセットアップ、ファイルヘッダーの読み込み、圧縮ディレクトリの読み込み、といった処理に余分な時間がかかります。その一方で、ファイルが分割不可能でサイズが非常に大きいときには、単一のリーダーがファイル全体の読み込みを完了するまで、クエリ自体の処理は行われません。この場合、並列性が下がってしまいます。

細切れファイル問題に対する解決法のひとつとして、EMR の S3DistCP ユーティリティがあります。これを使うことで、小さなファイル群を大きなオブジェクトへとまとめることができます。S3DistCP は、大量のデータを最適なやり方で HDFS から S3、S3 からS3 、S3 から HDFS へと移動させる際にも利用できます。

大きなファイルにまとめるのは、複数の利点があります:

• 高速な一覧表示
• S3 へのリクエスト数の削減
• 管理するメタデータの削減

例:

以下では、単一の大きなファイルのテーブルと、5000 個の小さなファイルのテーブルとで、クエリ実行時間を比較しています。データはテキストフォーマットで、サイズは 7GB です。

5. 列指向データの作成を最適化する

Apache Parquet と Apache ORC はポピュラーな列指向データフォーマットです。両者には列方向圧縮、さまざまなエンコーディング、データ型に合わせた圧縮、プレディケイトプッシュダウン（訳注: プレディケイトプッシュダウンとは、WHERE 句や GROUP BY 句などの処理を効率的に行うための手法です。例えば GROUP BY に対するプレディケイトプッシュダウンでは、各リーダーで読み込んだデータについて、全データで GROUP BY を行う前に、各ワーカー内であらかじめ GROUP BY をしておき、その結果を集約ワーカーに転送する、といったプロセスをとります。各ワーカーで先に集約を行うことでデータの転送コストが下がり、結果的にパフォーマンスが向上します。このように、クエリの実行パイプラインの最後で行う処理を、効率化のためにあらかじめ各プロセスでおこなっておく（= プッシュダウン）のが、プレディケイトプッシュダウンの役割となります）といった、データを効率的に持つための機能があります。両者はともに分割可能です。一般的に、高い圧縮率やデータブロックのスキップによって、S3 から読み込むデータが減り、その結果よりよいクエリパフォーマンスが得られます。

チューニング可能なパラメーターのひとつとして、ブロックサイズまたはストライプのサイズがあります。Parquet におけるブロックサイズ、 ORC におけるストライプサイズというのは、バイトサイズ単位で表されるもので、単一ブロックで保持できる最大行数を意味します。ブロックまたはストライプのサイズが大きいほど、単一ブロックで格納できる行数も多くなります。デフォルトでは Parquet のブロックサイズは 128MB で、ORC のストライプサイズは 64MB になります。テーブル内に大量のカラムがある場合は、各カラムのブロック単位で効率的にシーケンシャル I/O を行えるように、より大きなブロックサイズにすることをおすすめします。

そのほかのパラメーターとして、データブロックの圧縮アルゴリズムが挙げられます。Parquet のデフォルト圧縮フォーマットは Snappy ですが、それ以外に圧縮なし、 GZIP、そして LZO も使用可能です。ORC のデフォルトは ZLIB ですが、それ以外に圧縮なしと Snappy が利用可能です。デフォルトの圧縮アルゴリズムからはじめて、10GB 以上のデータサイズになった場合には、それ以外の圧縮アルゴリズムを試してみることをおすすめします。

Parquet/ORC ファイルフォーマットは、ともにプレディケイトプッシュダウン（プレディケイトフィルタリングとも呼ばれます）をサポートしています。Parquet/ORC はともに、ブロック内にカラムの値を保持するだけでなく、最大値/最小値のようなブロックごとの統計データも持っています。クエリが実行される際に、統計情報によって当該ブロックを読み込む必要があるか、スキップしてよいかを判断します。スキップするブロック数を最適化するための一つの方法として、Parquet や ORC で書き出す前に、よくフィルタされるカラムでデータをソートしておくやり方があります。これによって、各ブロックにおける最小値と最大値の差分が、全体としてもっとも小さくなることが保証されます。これによって、フィルタ効率を向上させることができます。

Amazon EMR 上で Spark や Hive を実行することで、既存のデータを Parquet や ORC に変換できます。 詳細については、S3のデータをAmazon Athenaを使って分析する のブログポストを参照してください。また以下のリソースもあります:

• https://aws.amazon.com/jp/blogs/news/build-a-data-lake-foundation-with-aws-glue-and-amazon-s3/ のブログ記事
• aws-blog-spark-parquet-conversion Spark GitHub リポジトリ
• Converting to Columnar Formats (hive によるフォーマットの変換)

おまけのテクニック

Apache Parquet とApache ORC は Struct 型のデータを、ファイル内にデータブロックとして保持します。クエリの対象に Struct 型のカラムが含まれている場合、データブロックに含まれるすべての構造化データを読み取った上で、クエリエンジン側でデシリアライズする必要があります。もしStruct のサイズが大きかったら、この処理は非常に重たくなります。もしクエリの中で必要なフィールドが，構造化データ内のごく限られたフィールドなのであれば、そのフィールドだけを抜き出して別のカラムとして持つことによって、この重たい前構造化データの読み取りを避けることができます。つまり、不必要な読み取り、データのデシリアライズを行わなくてよくなります。

例えば、以下のようなテーブルがあったとします。

CREATE EXTERNAL TABLE products (
	product_id INT,
	product_info struct<name STRING: description STRING, dimensions STRING: industry STRING>,
	cost DOUBLE
) …

もし product_info.name に対して頻繁にクエリを投げるのであれば、当該カラムを Struct から抜き出すのが良いでしょう。より最適化したカラム構成は以下の通りです。

CREATE EXTERNAL TABLE products (
	product_id INT,
	name STRING,
	product_info struct<description STRING, dimensions STRING: industry STRING >,
	cost DOUBLE
) …

例:

データセット: 32.GB、Parquet フォーマット, ~600M 行

SELECT count(*) FROM lineitem where l_returnflag like '%s'

SELECT count(*) FROM lineitem where data.l_returnflag like '%s'

構造化データを AWS Glue を使って展開するやり方に関しては、AWS Glue : ネストされた JSON を Relationalizeトランスフォームをご覧ください。

クエリチューニング

Athena ではクエリの実行エンジンとして Presto を使用しています。クエリ実行時に Presto がどのように動いているか理解することで、クエリの最適化方法について把握することができます。

このセクションでは、以下のベストプラクティスについて説明します。

• ORDER BY を最適化する
• JOIN を最適化する
• GROUP BY を最適化する
• LIKE 演算子を最適化する
• 必要なカラムだけを読みこむ

6. ORDER BY を最適化する

ORDER BY 句はクエリの実行結果をソートして返します。ソートを実行するために、Presto はデータのすべてのレコードを単一のワーカーに送り、それからソートを実行します。この処理は Presto のメモリを大量に消費するため、クエリの実行時間が非常に長くなります。最悪の場合には、クエリが失敗します。

トップ N の値をみるために ORDER BY 句を使用する場合には、単一ワーカーでソートを実行するのではなく、LIMIT 句によって各ワーカーにソートとリミットの処理をプッシュし、ソートにかかるコストを大きく削減してください。

例:

データセット: 7.25GB、無圧縮、テキストフォーマット、6000万行

7. JOIN を最適化する

2 つのテーブルを結合する際には、両者のうち大きなほうを左側に、小さなほうを右側に指定してください。Presto は JOIN 句の右側で指定されたテーブルを各ワーカーノードに転送して、左側のテーブルを順になめていくことで結合を行います。右側のテーブルを小さくすることで、メモリ消費量を少なく、クエリを高速に実行することができます。

例:

データセット: 74GB、無圧縮、テキストフォーマット、6億200万行

例外として、複数のテーブルをまとめて結合する場合と、クロス結合が行われる場合があります。Presto は結合の実行順序の最適化をサポートしていないため、が左側から右側に対して結合処理が行われます。そのためテーブルを左側から大きい順に並べることにより、結合条件に合わせたテーブルの並びにならない場合、クロス結合が実行されてしまいます（訳注: 以下の例の上側のクエリの場合、まず lineitem と customer とで結合処理が行われます。しかし結合条件にあるのは lineitem と orders、ccustomer と orders のため、lineitem と customer の間ではクロス結合が行われてしまいます。その結果得られた巨大なテーブルと orders の間で，次の結合処理が行われます。クロス結合は非常にコストの高い処理のため、この場合は 30 分以内にクエリが完了せず、タイムアウトしてしまっています）。

例:

データセット: 9.1GB、無圧縮、テキストフォーマット、7600万行

8. GROUP BY を最適化する

GROUP BY 演算子は、指定されたカラムの値に応じてレコードを各ワーカーノードに分散し、各ノードはメモリ内に対象の値を保持します。レコードが追加されると、メモリ内の GROUP BY 対象のカラムを比較します。一致した場合には、対象となる値を集約します。

クエリ内で GROUP BY を使う際には、カーディナリティ（訳注: カラム内のユニークな値の個数）が高い順にカラムを並べてください（これはユニークな値の数が多いほど、データが各ワーカーに均等に分散するためです）。

SELECT state, gender, count(*) 
           FROM census 
GROUP BY state, gender;

もうひとつの最適化方法は、可能ならば GROUP BY の対象を文字列でなく数字にすることです。数字は文字列に比べて必要とするメモリが少ないため、高速に処理することができます。

またその他の方法として、SELECT の中で扱うカラム数を制限することによって、レコードをメモリに確保したり、GROUP BY で集約する際に必要とするメモリ総量を減らすやり方があります。

9. 近似関数を使う

大規模なデータセットを処理する際の典型的なユースケースとして、COUNT(DISTINCT column) を使って、特定のカラムのユニークな値ごとのレコード数を求めるというものがあります。例として挙げられるのは、ウェブサイトを訪れたユニークユーザーを求めるというものです。

もし正確な値が必要ない場合、例えばサイト内のどのウェブページを突っ込んで調べるべきか判断するといったケースでは、approx_distinct() の使用を検討してみてください。この関数は、全文字列を捜査するかわりに、ユニークなハッシュの数をカウントすることで、メモリ使用量を最小限におさえます。この手法の欠点は、得られた値に 2.3% の標準誤差を持つことです。

例:

データセット: 74GB、無圧縮、テキストフォーマット、6億行

詳しくは Presto ドキュメントの Aggregate Functions を参照してください。

10. 必要なカラムだけを読み込む

クエリを実行する際に、すべてのカラムを使用するかわりに、最後の SELECT 句で必要なカラムだけに絞ってください。処理対象のカラム数を減らすことで、クエリの実行パイプライン全体で処理しなければいけないデータ総量を削減できます。これは特に、大量のカラムがあるテーブルや、文字列ベースのカラムが主体のテーブルにクエリを投げるときに有効です。

例:

データセット: 7.25GB、無圧縮、テキストフォーマット、6000万行

おまけ: パーティション処理の最適化

テーブルに非常に多くのパーティションがある場合、パーティション情報の処理が Athena のクエリにおけるボトルネックとなる場合があります。このオーバーヘッドを小さくするための方法がふたつあります。ひとつめのアプローチは、各パーティションカラムに対して明示的にひとつを指定して、単一のパーティションを対象とした処理を行う場合に有効です。ふたつめのアプローチは、パーティションプロジェクション（Partition Projection）を用いて、メタストアからパーティション情報を取り出さずに、パーティション情報を計算してクエリを実施するやり方です。

アプローチ 1: 単一パーティションへのクエリ

パーティションを持つテーブルにクエリを投げる場合、Athena の最適化処理によってクエリ実行時間を大きく削減できます。WHERE 句で単一パーティションを明示的に指定する場合、パーティション情報の処理をスキップしてクエリを実行します。この最適化処理により、テーブルに含まれるパーティション数にかかわらず、クエリは一定時間でパーティション情報を取得することができます。この最適化処理は、パーティションカラムのデータ型が STRING 形のときにのみ適用されます。もし複数のパーティションカラムのうちひとつでも STRING 形でないものが含まれるか、すべてのパーティションカラムが指定されなかった場合、Athena はこの最適化処理を利用できません。

*このテーブルのパーティションカラムは l_orderdate です

アプローチ 2

パーティションプロジェクションを用いて、大量のパーティションを持つテーブルにおけるパーティション管理を自動化し、Athena のクエリを高速化できます。

パーティションプロジェクションでは、AWS Glue Data Catalog を用いずに、テーブルの設定情報からパーティションの値と場所を計算します。離れた場所にアクセスするのと比べてメモリ内処理は高速なことから、パーティションプロジェクションは特に大量のパーティションを持つテーブルにおいてクエリ実行時間の削減につながります（訳注: この場合は Glue Data Catalog とアクセスするのと比較して、メモリ内で設定情報からパーティション情報を計算する方が高速ということです）。クエリおよび保存されているデータによりますが、パーティションのメタデータを取得する場合と比べて、パーティションプロジェクションは大きなクエリ実行時間に削減につながります。

テーブルに含まれるパーティションのスキーマが同一であるか、テーブルスキーマが常にパーティションスキーマと連動する形であるならば、パーティションプロジェクションを使うのに適しています。ID や詳細な日付情報のような、カーディナリティの非常に高いカラムをパーティションとして使うのが、この例として当てはまります。

詳細情報については、Partition Projection with Amazon Athena を参照ください。

結論

このポストでは、Amazon Athena および Presto エンジンにおけるインタラクティブ分析を最適化するための、ベストプラクティスについて述べました。これらは（訳注: パーティションプロジェクションを除き）Amazon EMR 上で Presto を動かす際にも、同様に適用可能です。

著者について

Manjeet Chayel は AWS のソリューションアーキテクトです。
Mert Hocanin is は Amazon EMR をメインとした ビッグデータのアーキテクトです。

原文: Top 10 Performance Tuning Tips for Amazon Athena (翻訳: SA志村)