Icebergのパーティション変更した場合のS3ディレクトリ構造の変化を確認する

Last updated at 2024-08-16Posted at 2024-08-16

確認したいこと

Iceberg は、パーティションを途中で変更することができます。

物理データをS3に格納した場合、パーティションを変更したら物理的なS3のディレクトリ構造はどう変化するのかを確認してみました。

結論

変更前はdt_day（日付）でパーティションを切って、変更後はidでパーティションを切った場合は以下のようになりました。

変更前

myfolder
└── data
    ├── 2pM-Cg
    │   └── dt_day=2024-08-01
    ├── puBT4w
    │   └── dt_day=2024-08-02
    └── Mte4cw
        └── dt_day=2024-08-03

変更後

myfolder
└── data
    ├── 2pM-Cg
    │   └── dt_day=2024-08-01
    ├── puBT4w
    │   └── dt_day=2024-08-02
    ├── Mte4cw
    │   └── dt_day=2024-08-03
    ├── fs5Tg
    │   └── id=4
    └── gH2-w
        └── id=5

dataフォルダの下にランダム英数字のフォルダがあり、その下にパーティションが存在します。そのパーティションが、途中から変わりました。

やってみる

AthenaでIcebergテーブルを作成する

以下を参考に、AthenaでCREATE文を実行しました。

CREATE TABLE default.iceberg_test_tbl (id bigint, data string, dt timestamp)
  PARTITIONED BY (day(dt))
  LOCATION 's3://my-bucket/tbl1-test/'
  TBLPROPERTIES ( 'table_type' = 'ICEBERG' );

パーティションもうまく設定されています。

Athenaでレコード追加する

以下の5件を追加しました。
ちなみに、1行ずつ選択して実行していきました（一気に実行できないため）。

INSERT INTO iceberg_test_tbl VALUES (1,'test',TIMESTAMP '2024-08-01 01:12:13');
INSERT INTO iceberg_test_tbl VALUES (2,'test',TIMESTAMP '2024-08-01 02:12:13');
INSERT INTO iceberg_test_tbl VALUES (3,'test',TIMESTAMP '2024-08-02 03:12:13');
INSERT INTO iceberg_test_tbl VALUES (4,'test',TIMESTAMP '2024-08-02 04:12:13');
INSERT INTO iceberg_test_tbl VALUES (5,'test',TIMESTAMP '2024-08-03 05:12:13');

タイムスタンプは、単純にシングルクォートで囲っただけだと文字列として認識されてしまい失敗します。
Athenaのデータ型の例を参考に、上記のように指定しました。

SELECT文を実行し、上手く追加されていることを確認します。

S3のディレクトリを確認する

以下のように、5つのランダム英数字のフォルダが作成されています。5件レコード追加したからでしょう。

1つ中身を見てみると、このようにdt_dayと日付でパーティションが切られています。

Glueでパーティションを変更する

Athenaではパーティション変更するDDLはサポートしていないため（2024年8月16日時点）、Glueで実行していきます。

Glueで以下のジョブ（Glueのバージョンは4）を実行しました。SQL文はドキュメントの以下のあたりを参考にしました。

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

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

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

# パーティション削除
spark.sql("ALTER TABLE glue_catalog.default.iceberg_test_tbl DROP PARTITION FIELD dt_day")

# パーティション追加
spark.sql("ALTER TABLE glue_catalog.default.iceberg_test_tbl ADD PARTITION FIELD id")

job.commit()

GlueでIcebergにクエリを実行する際は、以下に記載のあるようにジョブパラメータを設定するのを忘れないようにしましょう。

--confのValueは、改行も入るかなり長い値です。

Athenaでパーティションを確認すると、変更されていますね。

Athenaでレコード追加する（変更後）

追加で2件のレコードを追加しました。

INSERT INTO iceberg_test_tbl VALUES (11,'test',TIMESTAMP '2024-08-04 06:12:13');
INSERT INTO iceberg_test_tbl VALUES (12,'test',TIMESTAMP '2024-08-05 07:12:13');