0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@naritomo08(nari tomo)

【第7編】Amazon S3のrawログとApache Icebergに保持期間とメンテナンスを設定する

0
Last updated at Posted at 2026-09-20

はじめに

HDFSからAmazon S3への移行後も、保存先を変更しただけではdataは増え続ける。本記事では、S3 rawを7日、batch Icebergを14日、Flink realtime Icebergを当日分だけ保持する。

保持期間と物理file削除は分けて扱う。

対象 table / prefix data保持期間 削除方法
Raw raw/syslog/、raw/authlog/ 7日 S3 Lifecycle
batch Iceberg syslog_iceberg、authlog_iceberg 当日を含む14日 Iceberg DELETE
Flink event syslog_events、authlog_events 当日 Iceberg DELETE
Flink 1分集計 syslog_host_1m、authlog_host_1m 当日 Iceberg DELETE

Iceberg table配下へS3 Lifecycleやaws s3 rmを設定しない。Icebergはdata fileだけでなくsnapshot、manifest、metadataの参照関係を持つため、削除とmaintenanceはIceberg procedureから実行する。

前提

  • S3 rawとS3 Icebergへの移行が完了している
  • Glue Databaseはlogsである
  • Sparkからglue_prod catalogを利用できる
  • dt列はdate型である
  • Flink jobのcheckpointとIceberg commitが正常である
  • 操作端末にAWS CLI v2とjqがある

環境を読み込む。

export AWS_REGION=ap-northeast-1

source /etc/rawlog-s3-backup/common.env
export RAW_BUCKET="${S3_BUCKET_NAME}"

source /etc/iceberg/aws.env
export LAKE_BUCKET="${ICEBERG_BUCKET}"
export GLUE_DATABASE="${GLUE_DATABASE:-logs}"

export AWS_RAW_PROFILE=home-rawlog-backup
export AWS_ADMIN_PROFILE=home-aws-admin

aws sts get-caller-identity --profile "${AWS_ADMIN_PROFILE}"
aws s3api head-bucket --bucket "${RAW_BUCKET}" --profile "${AWS_RAW_PROFILE}"
aws glue get-tables \
  --region "${AWS_REGION}" \
  --database-name "${GLUE_DATABASE}" \
  --query 'TableList[].Name' \
  --profile "${AWS_ADMIN_PROFILE}"

tableをbatchとrealtimeへ分類する

本構成では用途ごとにtableを分離している。

syslog_eventsとauthlog_eventsはevent明細、_host_1mはFlinkが生成する1分集計である。いずれもrealtime系として当日分だけ保持する。

Rawへ7日のS3 Lifecycleを設定する

まず既存Lifecycleをbackupする。put-bucket-lifecycle-configurationはbucketの既存設定全体を置き換えるため、他のruleがある場合は必ず同じJSONへ残す。

aws s3api get-bucket-lifecycle-configuration \
  --bucket "${RAW_BUCKET}" \
  --profile "${AWS_ADMIN_PROFILE}" \
  > raw-lifecycle-before.json

Lifecycle設定へ次の2 ruleを追加する。

{
  "Rules": [
    {
      "ID": "expire-raw-syslog-after-7-days",
      "Status": "Enabled",
      "Filter": {"Prefix": "raw/syslog/"},
      "Expiration": {"Days": 7},
      "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 1}
    },
    {
      "ID": "expire-raw-authlog-after-7-days",
      "Status": "Enabled",
      "Filter": {"Prefix": "raw/authlog/"},
      "Expiration": {"Days": 7},
      "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 1}
    }
  ]
}

上記をraw-lifecycle.jsonとして保存し、既存ruleがある場合はそのruleもRulesへ含めて適用する。

aws s3api put-bucket-lifecycle-configuration \
  --bucket "${RAW_BUCKET}" \
  --lifecycle-configuration file://raw-lifecycle.json \
  --profile "${AWS_ADMIN_PROFILE}"

aws s3api get-bucket-lifecycle-configuration \
  --bucket "${RAW_BUCKET}" \
  --profile "${AWS_ADMIN_PROFILE}" \
  | jq '.Rules[] | {ID, Status, Filter, Expiration}'

Expiration.Days=7はobject作成時刻を基準にする。directory名の日付を基準に削除する設定ではない。再投入が必要なdataは、objectがexpireする前に処理する。

Iceberg maintenance jobを作成する

毎日1回実行するPySpark jobを用意する。次の順序を維持する。

  1. 保持期間外のrowをDELETEする
  2. small fileをcompactionする
  3. 古いsnapshotをexpireする
  4. orphan file候補をdry-runで確認する
  5. 十分に古いorphan fileだけ削除する
# iceberg_retention_maintenance.py
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from pyspark.sql import SparkSession

CATALOG = "glue_prod"
DATABASE = "logs"

BATCH_TABLES = (
    "syslog_iceberg",
    "authlog_iceberg",
)

REALTIME_TABLES = (
    "syslog_events",
    "authlog_events",
    "syslog_host_1m",
    "authlog_host_1m",
)

spark = SparkSession.builder.appName("iceberg-retention-maintenance").getOrCreate()

today = datetime.now(ZoneInfo("Asia/Tokyo")).date()
# 今日を含めて14 calendar daysを残す。
batch_cutoff = today - timedelta(days=13)
realtime_cutoff = today

def full_name(table):
    return f"{CATALOG}.{DATABASE}.{table}"

def delete_old_rows(table, cutoff):
    spark.sql(
        f"DELETE FROM {full_name(table)} "
        f"WHERE dt < DATE '{cutoff.isoformat()}'"
    )

def maintain(table, snapshot_days, retain_last):
    procedure_table = f"{DATABASE}.{table}"

    # DELETE後に残ったsmall fileをまとめる。
    spark.sql(f"""
      CALL {CATALOG}.system.rewrite_data_files(
        table => '{procedure_table}',
        strategy => 'binpack',
        options => map(
          'target-file-size-bytes', '536870912',
          'min-input-files', '5'
        )
      )
    """).show(truncate=False)

    spark.sql(f"""
      CALL {CATALOG}.system.expire_snapshots(
        table => '{procedure_table}',
        older_than => TIMESTAMP '{(today - timedelta(days=snapshot_days)).isoformat()} 00:00:00',
        retain_last => {retain_last}
      )
    """).show(truncate=False)

    # Flinkが利用中のfileを誤削除しないよう、3日より新しいfileは対象外にする。
    orphan_before = today - timedelta(days=3)
    spark.sql(f"""
      CALL {CATALOG}.system.remove_orphan_files(
        table => '{procedure_table}',
        older_than => TIMESTAMP '{orphan_before.isoformat()} 00:00:00',
        dry_run => true,
        stream_results => true
      )
    """).show(100, truncate=False)

    spark.sql(f"""
      CALL {CATALOG}.system.remove_orphan_files(
        table => '{procedure_table}',
        older_than => TIMESTAMP '{orphan_before.isoformat()} 00:00:00',
        dry_run => false,
        stream_results => true
      )
    """).show(truncate=False)

for table in BATCH_TABLES:
    delete_old_rows(table, batch_cutoff)
    maintain(table, snapshot_days=7, retain_last=10)

for table in REALTIME_TABLES:
    delete_old_rows(table, realtime_cutoff)
    # dataは当日分だけにするが、Flink復旧用snapshotは3日かつ最低100件残す。
    maintain(table, snapshot_days=3, retain_last=100)

spark.stop()

「realtimeは当日」という条件はtableから参照できるrowの保持期間であり、snapshotとorphan fileを即日物理削除する指定ではない。Flinkのcheckpoint復旧に必要なsnapshotと一時fileを保護するため、snapshotは3日、orphan fileは最低3日残す。

初回は削除対象を確認する

初回実行前に件数と日付範囲を記録する。

SELECT 'syslog_iceberg', COUNT(*), MIN(dt), MAX(dt) FROM glue_prod.logs.syslog_iceberg
UNION ALL
SELECT 'authlog_iceberg', COUNT(*), MIN(dt), MAX(dt) FROM glue_prod.logs.authlog_iceberg
UNION ALL
SELECT 'syslog_events', COUNT(*), MIN(dt), MAX(dt) FROM glue_prod.logs.syslog_events
UNION ALL
SELECT 'authlog_events', COUNT(*), MIN(dt), MAX(dt) FROM glue_prod.logs.authlog_events
UNION ALL
SELECT 'syslog_host_1m', COUNT(*), MIN(dt), MAX(dt) FROM glue_prod.logs.syslog_host_1m
UNION ALL
SELECT 'authlog_host_1m', COUNT(*), MIN(dt), MAX(dt) FROM glue_prod.logs.authlog_host_1m;

本番tableへ適用する前に、job内のDELETEとdry_run => falseをcomment outして、compaction対象とorphan候補を確認する。特に初回のorphan候補に直近のFlink data fileが含まれる場合は実行しない。

実行する

既存のSpark Iceberg実行環境から起動する例を示す。

sudo -u spark /usr/local/spark/bin/spark-submit \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.glue_prod=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.glue_prod.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog \
  --conf spark.sql.catalog.glue_prod.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
  --conf "spark.sql.catalog.glue_prod.warehouse=s3://${LAKE_BUCKET}/warehouse" \
  iceberg_retention_maintenance.py

Flink realtime tableへのmaintenanceは、checkpoint成功直後かつjobが安定している時間に実行する。Flink jobが再起動中、checkpoint失敗中、savepoint復旧中の場合は実行しない。

定期実行する

毎日01:30に実行する例である。日次batchやFlinkの再起動時間と重ならない時刻へ調整する。

# /etc/systemd/system/iceberg-retention-maintenance.service
[Unit]
Description=Iceberg retention and maintenance

[Service]
Type=oneshot
User=spark
EnvironmentFile=/etc/iceberg/aws.env
ExecStart=/usr/local/bin/run-iceberg-retention-maintenance

# /etc/systemd/system/iceberg-retention-maintenance.timer
[Unit]
Description=Run Iceberg retention maintenance daily

[Timer]
OnCalendar=*-*-* 01:30:00 Asia/Tokyo
Persistent=true
RandomizedDelaySec=300

[Install]
WantedBy=timers.target

sudo systemctl daemon-reload
sudo systemctl enable --now iceberg-retention-maintenance.timer
systemctl list-timers iceberg-retention-maintenance.timer

AWSマネージドサービス移行後は、同じPySpark fileをEMR Serverlessへ配置し、EventBridge Schedulerなどから1日1回起動する。local AWS profileはjobへ渡さず、runtime roleへGlueと対象Iceberg prefixのread/write/delete権限を付与する。

実行後に確認する

保持期間を確認する。

SELECT MIN(dt), MAX(dt), COUNT(*) FROM glue_prod.logs.syslog_iceberg;
SELECT MIN(dt), MAX(dt), COUNT(*) FROM glue_prod.logs.authlog_iceberg;
SELECT MIN(dt), MAX(dt), COUNT(*) FROM glue_prod.logs.syslog_events;
SELECT MIN(dt), MAX(dt), COUNT(*) FROM glue_prod.logs.authlog_events;
SELECT MIN(dt), MAX(dt), COUNT(*) FROM glue_prod.logs.syslog_host_1m;
SELECT MIN(dt), MAX(dt), COUNT(*) FROM glue_prod.logs.authlog_host_1m;

snapshotとfile数も確認する。

SELECT committed_at, snapshot_id, operation
FROM glue_prod.logs.syslog_iceberg.snapshots
ORDER BY committed_at DESC;

SELECT COUNT(*) AS data_files,
       CAST(AVG(file_size_in_bytes) AS BIGINT) AS avg_file_size
FROM glue_prod.logs.syslog_iceberg.files;

次も監視する。

  • maintenance jobの成功時刻と所要時間
  • DELETE、compaction、snapshot expirationで削除された件数
  • Flink checkpointの成功時刻
  • Kafkaコンシューマーの遅延
  • Glueの最新metadata_location
  • S3 storage sizeとobject数
  • Athenaから見た最古／最新のdt

rollback

S3 Lifecycleは即時実行ではないが、expire済みobjectはversioningや別backupがなければ復元できない。誤設定時は、backupしておいた既存Lifecycleを戻す。

aws s3api put-bucket-lifecycle-configuration \
  --bucket "${RAW_BUCKET}" \
  --lifecycle-configuration file://raw-lifecycle-before.json \
  --profile "${AWS_ADMIN_PROFILE}"

IcebergのDELETEはsnapshot expiration前なら、対象snapshotを確認してrollbackできる。snapshot expiration後やorphan file削除後は復旧できない場合があるため、初回は削除procedureを無効化し、複数回の定期実行結果を確認してから有効化する。

参考資料

関連記事

実機検証について

本手順は執筆時点では一連の実機検証を完了していない。最初は検証tableと短い対象期間で、件数、snapshot、Flink checkpoint、orphan候補を確認してから本番tableへ適用する。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?