AWS AthenaからDelta lakeを直接読む

Delta Lake内のデータは、SparkやDatabricks SQLのエンドポイント(ODBC/JDBC)などを経由してBIをはじめとした外部のさまざまなサービスから参照することができます。一方で、ツールの制限から直接こうしたエンドポイントに接続できない場合もあります(Amazon QuickSightなど)。このような場合には、AWS Athenaなどのサービスを経由して接続する形態になります。

(アップデート 2022年11月22日)
AWS QuickSightがDatabricks(Databricks SQL)をサポートしました。今後は、QuickSightからDeltaLakeを参照する場合、Athenaを経由せず、直接Databricks SQLのエンドポイントに接続することができます。詳しくは以下の記事を参照ください。

この記事では、Delta LakeのテーブルデータにAWS Athenaから接続する方法について説明します。Delta Lake自体はオープンソースなのであらゆる環境で動作するのですが、ここではDatabricksを想定します。その他の環境についてもほぼ同様の方法で実現可能です。

仕組み

Delta Lakeの実体は、

• メタデータ管理のためのDeltaログファイル(JSON)
• 実データ保持のParquetファイル群

の2種類のファイルで構成されます。また、多くの場合、AWS S3などのオブジェクトストレージ上に保管されています。よって、AthenaからこのS3上にアクセスすることで、Delta Lakeのテーブルが参照できます。

手順

以下の通りの手順で進めていきます。

1. Delta Lakeデータの用意
2. Delta Lakeからマニフェストファイルを作成
3. AWS Athena上で外部テーブルを作成
4. AWS Athenaからクエリを実行
5. Delta Lakeを更新時にはマニフェストファイルを更新

1. Delta Lakeデータの用意

以下のDelta LakeのテーブルをAWS Athenaから参照する場合を見ていきます。

• Delta Lakeのデータパス: s3://mydelta/foo/lending-club-loan/

以下のファイル構成になっています(例)。

### Databricks Notebook上

%fs ls s3://mydelta/foo/lending-club-loan/

_delta_log/                    
part-00000-2094d023-de99-414f-9288-c5ce70210736-c000.snappy.parquet
part-00000-5643b790-058f-4953-9e7e-86a79b3b89f1-c000.snappy.parquet
part-00000-b3071114-b0ab-42f5-a863-7d77a1b3f483-c000.snappy.parquet
...

2. Delta Lakeからマニフェストファイルを作成

Databricks上から以下のコマンドを実行し、マニフェストファイルを作成します。

### Databricks Notebook上

%sql

GENERATE symlink_format_manifest FOR TABLE delta.`s3a://mydelta/foo/lending-club-loan/`

マニフェストファイルの中身はテーブルを構成するParquetファイルのリストになっています。上の図にある通り、ここで生成するマニフェストファイルはDeltaファイルの同じディレクトリ内に出力されます。

確認してみます。

### Databricks Notebook上

%fs ls s3://mydelta/foo/lending-club-loan/

_delta_log/
_symlink_format_manifest/  # <==　マニフェストファイル(ディレクトリ)             
part-00000-2094d023-de99-414f-9288-c5ce70210736-c000.snappy.parquet
part-00000-5643b790-058f-4953-9e7e-86a79b3b89f1-c000.snappy.parquet
part-00000-b3071114-b0ab-42f5-a863-7d77a1b3f483-c000.snappy.parquet
...

また、この後、Athenaのテーブルを作成する際に、このDelta LakeのSchemaを入力するので、ここで確認しておきます。

### Databricks Notebook上

(
  spark.read.format('delta')
  .load('s3://mydelta/foo/lending-club-loan/')
  .printSchema()
)

[結果]
root
 |-- loan_amnt: float (nullable = true)
 |-- term: string (nullable = true)
 |-- int_rate: float (nullable = true)
 |-- grade: string (nullable = true)
 |-- state: string (nullable = true)
 |-- emp_title: string (nullable = true)
 |-- home_ownership: string (nullable = true)
 |-- annual_inc: float (nullable = true)
 |-- loan_status: string (nullable = true)

3. AWS Athena上で外部テーブルを作成

Athena上で外部テーブル(lending_club)の作成クエリを実行します。このクエリの中で、先ほど確認したテーブルのSchemaをマニュアル入力する必要があります。また、Delta LakeでPartitionがある場合、補足を参照ください

--- AWS Athenaのクエリ

CREATE EXTERNAL TABLE `lending_club` (
  loan_amnt float,
  term string,
  int_rate float,
  grade string,
  state string,
  emp_title string,
  home_ownership string,
  annual_inc float,
  loan_status string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://mydelta/foo/lending-club-loan/_symlink_format_manifest/'

4. AWS Athenaからクエリを実行

テーブルが作成できましたので、後は通常通りAthena上でSQLを実行します。

5. Delta Lakeを更新時にはマニフェストファイルを更新

Delta Lakeはテーブルを更新するたびに、構成するParquetファイルが変わります。そのため、その度にマニフェストファイルを更新する必要があります。

マニュアルでは上で実施したGENERATEクエリを実行しても良いのですが、煩わしいので、自動的にマニフェスをと更新するテーブルプロパティが用意されています。以下のクエリにより、Delta Lakeのテーブルが更新されるとそれに伴ってマニフェストファイルも更新されます。

### Databricks Notebook上

%sql

ALTER TABLE delta.`s3://mydelta/foo/lending-club-loan/` SET TBLPROPERTIES(delta.compatibility.symlinkFormatManifest.enabled=true)

参考

• Presto and Athena to Delta Lake integration

補足: Delta LakeでのPartitionがある場合

Delta LakeでPartitionしている場合は、Athenaの外部テーブル作成時にそれを伝える必要があります。

--- AWS Athenaのクエリ

CREATE EXTERNAL TABLE mytable ([(col_name1 col_datatype1, ...)])
[PARTITIONED BY (col_name2 col_datatype2, ...)] -- <==ここでPartitionを伝える
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION '<path-to-delta-table>/_symlink_format_manifest/'