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)

【第4編】Apache FlinkからAWS S3上のApache Icebergへリアルタイムに書き込んでみる

0
Last updated at Posted at 2026-09-19

はじめに

これまで、KafkaのsyslogとauthlogをApache Flinkで読み、HDFS上のApache Icebergへリアルタイムに書き込んでいた。

また、別の記事ではHDFSまたはローカルS3上のIcebergを、Sparkのbatch処理でAWS S3へコピーした。

今回はbatchコピーではなく、FlinkからAWS Glue Data CatalogとAWS S3上のIcebergへ直接書き込む。

hdfsex-flink-to-aws-iceberg.png

作成するtableは次の通りである。

glue_prod.logs.syslog_events
glue_prod.logs.authlog_events

batch用のsyslog_iceberg、authlog_icebergとはtableを分け、既存のリアルタイムtable名を引き継ぐ。

batch同期との違い

方式 用途 commit単位
Spark batch 過去データ移行、日次再作成 日付単位
Flink streaming Kafkaから到着する新規データ checkpoint単位

同じIceberg tableへSparkとFlinkが同時に書き込むこともできるが、commit競合、再処理範囲、運用責任が分かりにくくなる。最初はtableを分離して動作を確認する。

前提環境

項目 値
Flink 2.0.1
Java 17
Iceberg 1.10.1
Kafka broker kafka1:9092,kafka2:9092,kafka3:9092
Kafka topic syslog、authlog
AWS region ap-northeast-1
Glue database logs
Iceberg warehouse s3://<YOUR_ICEBERG_BUCKET>/warehouse
Flink構成 standalone 3 node

Flink nodeはflink1、flink2、flink3とする。3台すべてからKafka、AWS Glue、AWS S3へ接続できる必要がある。

raw JSONは次の形式を想定する。

{
  "ts": "2026-09-16T12:34:56+09:00",
  "src_host": "raspi4-3",
  "program": "sshd",
  "message": "..."
}

AWS側を準備する

S3 bucketを確認する

export AWS_REGION='ap-northeast-1'
export AWS_PROFILE='home-aws'
source /etc/iceberg/aws.env

aws sts get-caller-identity \
  --profile "${AWS_PROFILE}" \
  --region "${AWS_REGION}"

Iceberg構築記事で作成した既存bucketを継続利用する。新しいbucketは作成せず、設定を変更する必要がある場合だけhome-aws-adminへ切り替える。通常の確認とFlink実行用credentialにはhome-awsを使う。

aws s3api put-public-access-block \
  --bucket "${ICEBERG_BUCKET}" \
  --public-access-block-configuration \
    BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true \
  --profile "${AWS_PROFILE}"

aws s3api put-bucket-encryption \
  --bucket "${ICEBERG_BUCKET}" \
  --server-side-encryption-configuration \
    '{"Rules":[{"ApplyServerSideEncryptionByDefault":{"SSEAlgorithm":"AES256"}}]}' \
  --profile "${AWS_PROFILE}"

aws s3api put-bucket-versioning \
  --bucket "${ICEBERG_BUCKET}" \
  --versioning-configuration Status=Enabled \
  --profile "${AWS_PROFILE}"

IAM権限

Flink用IAM principalには、少なくとも次の権限が必要である。

  • Iceberg warehouseとcheckpoint prefixに対するS3 List/Get/Put/Delete
  • Glue Database/Table/TableVersionの取得、作成、更新

検証環境では専用IAM userのprofileを使用できるが、本番ではEC2 instance profile、IAM Roles Anywhereなどの短期credential方式を使用する。長期access keyをSQLやGitへ記載しない。

FlinkへJARを追加する

各Flink nodeへ次を配置する。

JAR 用途
flink-sql-connector-kafka-4.0.1-2.0.jar Kafka source
iceberg-flink-runtime-2.0-1.10.1.jar Flink Iceberg runtime
iceberg-aws-bundle-1.10.1.jar Glue Catalog、S3FileIO、AWS SDK

flink1で取得する。

cd /opt/flink/current/lib

sudo -u flink curl -fLO \
  https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/4.0.1-2.0/flink-sql-connector-kafka-4.0.1-2.0.jar

sudo -u flink curl -fLO \
  https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-runtime-2.0/1.10.1/iceberg-flink-runtime-2.0-1.10.1.jar

sudo -u flink curl -fLO \
  https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-aws-bundle/1.10.1/iceberg-aws-bundle-1.10.1.jar

JARとclassを確認する。

jar tf /opt/flink/current/lib/iceberg-flink-runtime-2.0-1.10.1.jar \
  | grep 'org/apache/iceberg/flink/FlinkCatalogFactory.class'

jar tf /opt/flink/current/lib/iceberg-aws-bundle-1.10.1.jar \
  | grep -E 'org/apache/iceberg/aws/(glue/GlueCatalog|s3/S3FileIO).class'

3台構成では同じJARをflink2とflink3へ配布する。

rsync -av /opt/flink/current/lib/ flink2:/tmp/flink-lib/
rsync -av /opt/flink/current/lib/ flink3:/tmp/flink-lib/

各workerで反映する。

sudo rsync -av /tmp/flink-lib/ /opt/flink/current/lib/
sudo chown -R flink:flink /opt/flink/current/lib

すべてのnodeでJARのファイル名とSHA-256が一致することを確認する。異なるIceberg versionや個別AWS SDK JARを混在させない。

S3 checkpoint pluginを有効にする

Icebergのdata fileはS3FileIOが書き込む。一方、Flinkのcheckpoint/savepointをAWS S3へ保存するにはFlink S3 filesystem pluginが必要である。

Flink 2.0.1に同梱されるPresto S3 pluginをcheckpoint用に使う。

sudo mkdir -p /opt/flink/current/plugins/s3-fs-presto
sudo cp \
  /opt/flink/current/opt/flink-s3-fs-presto-2.0.1.jar \
  /opt/flink/current/plugins/s3-fs-presto/
sudo chown -R flink:flink /opt/flink/current/plugins/s3-fs-presto

3台すべてで実施する。checkpoint pathにはPresto pluginを明示するs3p://を使用する。

AWS credentialをFlinkへ渡す

検証用profile

bare metal検証環境でprofileを使用する例である。各nodeに同一pathで配置する。

sudo install -d -o flink -g flink -m 700 /var/lib/flink/.aws
sudo install -o flink -g flink -m 600 \
  /tmp/aws-credentials /var/lib/flink/.aws/credentials
sudo install -o flink -g flink -m 600 \
  /tmp/aws-config /var/lib/flink/.aws/config

/tmpの元ファイルは配置後に安全に削除する。

systemd環境変数

sudo tee /etc/flink/aws.env >/dev/null <<'EOF'
AWS_REGION=ap-northeast-1
AWS_DEFAULT_REGION=ap-northeast-1
AWS_PROFILE=home-aws
AWS_SHARED_CREDENTIALS_FILE=/var/lib/flink/.aws/credentials
AWS_CONFIG_FILE=/var/lib/flink/.aws/config
EOF

sudo chown root:flink /etc/flink/aws.env
sudo chmod 640 /etc/flink/aws.env

JobManagerとTaskManagerのsystemd unitへ追加する。

[Service]
EnvironmentFile=/etc/flink/aws.env

既存unitに合わせてdrop-inを編集し、再起動する。

sudo systemctl edit flink-jobmanager
sudo systemctl edit flink-taskmanager
sudo systemctl daemon-reload
sudo systemctl restart flink-jobmanager flink-taskmanager

実際のunit名が異なる場合は置き換える。JobManagerと全TaskManagerへ同じ設定が必要である。

SQL Clientを手動実行するときも同じEnvironmentFileを読むwrapperを作る。

sudo tee /usr/local/bin/flink-sql-aws >/dev/null <<'EOF'
#!/usr/bin/env bash
set -euo pipefail

set -a
. /etc/flink/aws.env
set +a

exec /opt/flink/current/bin/sql-client.sh embedded "$@"
EOF

sudo chmod 755 /usr/local/bin/flink-sql-aws

checkpointをAWS S3へ変更する

/etc/flink/conf/config.yamlへ設定する。

execution.checkpointing.interval: 5 min
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 30 s
execution.checkpointing.timeout: 10 min
execution.checkpointing.num-retained: 3
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
execution.checkpointing.storage: filesystem
execution.checkpointing.dir: s3p://<YOUR_ICEBERG_BUCKET>/flink/checkpoints/aws-iceberg
execution.checkpointing.savepoint-dir: s3p://<YOUR_ICEBERG_BUCKET>/flink/savepoints/aws-iceberg

s3.region: ap-northeast-1

checkpointはIceberg sinkのcommitとKafka offsetの復元に必要である。data fileがS3に存在しても、checkpointが成功しなければIceberg snapshotへcommitされない場合がある。

Glue Catalogの疎通を確認する

sudo tee /opt/flink/jobs/check_aws_glue.sql >/dev/null <<'EOF'
CREATE CATALOG glue_prod WITH (
  'type' = 'iceberg',
  'catalog-type' = 'glue',
  'warehouse' = 's3://<YOUR_ICEBERG_BUCKET>/warehouse',
  'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
  'client.region' = 'ap-northeast-1'
);

SHOW DATABASES IN glue_prod;
EOF

sudo chown flink:flink /opt/flink/jobs/check_aws_glue.sql
sudo -u flink /usr/local/bin/flink-sql-aws \
  -f /opt/flink/jobs/check_aws_glue.sql

SQL内のbucket名は実際の値へ置き換える。ClassNotFoundExceptionはJAR、Unable to load credentialsはSQL Clientとdaemonの環境変数を確認する。

AWS側Icebergテーブルを作成する

sudo tee /opt/flink/jobs/create_aws_iceberg_tables.sql >/dev/null <<'EOF'
CREATE CATALOG glue_prod WITH (
  'type' = 'iceberg',
  'catalog-type' = 'glue',
  'warehouse' = 's3://<YOUR_ICEBERG_BUCKET>/warehouse',
  'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
  'client.region' = 'ap-northeast-1'
);

CREATE DATABASE IF NOT EXISTS glue_prod.logs;

CREATE TABLE IF NOT EXISTS glue_prod.logs.syslog_events (
  ts_raw STRING,
  ts TIMESTAMP(6),
  host STRING,
  program STRING,
  msg STRING,
  dt DATE,
  hr INT
)
PARTITIONED BY (dt, host)
WITH (
  'format-version' = '2',
  'write.distribution-mode' = 'hash',
  'write.parquet.compression-codec' = 'zstd',
  'location' = 's3://<YOUR_ICEBERG_BUCKET>/warehouse/logs/syslog_events'
);

CREATE TABLE IF NOT EXISTS glue_prod.logs.authlog_events (
  ts_raw STRING,
  ts TIMESTAMP(6),
  host STRING,
  program STRING,
  msg STRING,
  dt DATE,
  hr INT
)
PARTITIONED BY (dt, host)
WITH (
  'format-version' = '2',
  'write.distribution-mode' = 'hash',
  'write.parquet.compression-codec' = 'zstd',
  'location' = 's3://<YOUR_ICEBERG_BUCKET>/warehouse/logs/authlog_events'
);
EOF

sudo -u flink /usr/local/bin/flink-sql-aws \
  -f /opt/flink/jobs/create_aws_iceberg_tables.sql

bucket名を置き換えて実行する。Glueでも確認する。

aws glue get-tables \
  --database-name logs \
  --region ap-northeast-1 \
  --profile home-aws \
  --query 'TableList[].Name'

KafkaからS3 Icebergへ書き込む

syslogとauthlogを1つのFlink jobとしてsubmitする。無限streamのINSERTを単純に並べず、STATEMENT SETを使う。

sudo tee /opt/flink/jobs/kafka_to_aws_iceberg.sql >/dev/null <<'EOF'
SET 'execution.runtime-mode' = 'streaming';
SET 'table.local-time-zone' = 'Asia/Tokyo';

CREATE CATALOG glue_prod WITH (
  'type' = 'iceberg',
  'catalog-type' = 'glue',
  'warehouse' = 's3://<YOUR_ICEBERG_BUCKET>/warehouse',
  'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
  'client.region' = 'ap-northeast-1'
);

CREATE TEMPORARY TABLE syslog_json (
  ts STRING,
  src_host STRING,
  program STRING,
  message STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'syslog',
  'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092,kafka3:9092',
  'properties.group.id' = 'flink-aws-syslog-events-v1',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true'
);

CREATE TEMPORARY TABLE authlog_json (
  ts STRING,
  src_host STRING,
  program STRING,
  message STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'authlog',
  'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092,kafka3:9092',
  'properties.group.id' = 'flink-aws-authlog-events-v1',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true'
);

STATEMENT SET BEGIN

INSERT INTO glue_prod.logs.syslog_events
SELECT
  ts AS ts_raw,
  CAST(TO_TIMESTAMP(SUBSTRING(ts, 1, 19), 'yyyy-MM-dd''T''HH:mm:ss') AS TIMESTAMP(6)),
  TRIM(src_host),
  program,
  message,
  CAST(SUBSTRING(ts, 1, 10) AS DATE),
  CAST(SUBSTRING(ts, 12, 2) AS INT)
FROM syslog_json
WHERE src_host IS NOT NULL
  AND TRIM(src_host) <> ''
  AND ts IS NOT NULL;

INSERT INTO glue_prod.logs.authlog_events
SELECT
  ts AS ts_raw,
  CAST(TO_TIMESTAMP(SUBSTRING(ts, 1, 19), 'yyyy-MM-dd''T''HH:mm:ss') AS TIMESTAMP(6)),
  TRIM(src_host),
  program,
  message,
  CAST(SUBSTRING(ts, 1, 10) AS DATE),
  CAST(SUBSTRING(ts, 12, 2) AS INT)
FROM authlog_json
WHERE src_host IS NOT NULL
  AND TRIM(src_host) <> ''
  AND ts IS NOT NULL;

END;
EOF

sudo -u flink /usr/local/bin/flink-sql-aws \
  -f /opt/flink/jobs/kafka_to_aws_iceberg.sql

SQL内のbucket名を置き換えてから実行する。初回にKafka保持範囲を取り込むためearliest-offsetを指定している。新規到着分だけなら初回submit前にlatest-offsetへ変更する。

job再起動時は新しいgroup IDで最初から起動せず、checkpointまたはsavepointから復元する。

動作確認

Flink UI

次を確認する。

  • jobがRUNNING
  • syslogとauthlogの両sourceがrecordを読んでいる
  • checkpointが継続してCOMPLETED
  • failed checkpointが増えていない
  • backpressureが継続していない

job submit直後だけで判断せず、最低1回のcheckpoint完了を待つ。

S3とGlue

aws s3 ls \
  s3://<YOUR_ICEBERG_BUCKET>/warehouse/logs/syslog_events/ \
  --recursive --profile home-aws

aws s3 ls \
  s3://<YOUR_ICEBERG_BUCKET>/warehouse/logs/authlog_events/ \
  --recursive --profile home-aws

data/とmetadata/が生成されることを確認する。

Athena

Athena engine version 3から確認する。

SELECT
  dt,
  hr,
  COUNT(*) AS cnt,
  MIN(ts) AS min_ts,
  MAX(ts) AS max_ts
FROM logs.syslog_events
GROUP BY dt, hr
ORDER BY dt DESC, hr DESC
LIMIT 24;

SELECT host, program, COUNT(*) AS cnt
FROM logs.authlog_events
WHERE dt = current_date
GROUP BY host, program
ORDER BY cnt DESC;

savepointと再起動

sudo -u flink /opt/flink/current/bin/flink list

export FLINK_JOB_ID="${FLINK_JOB_ID:?set the running Flink job ID}"
export ICEBERG_BUCKET="${ICEBERG_BUCKET:?set the AWS Iceberg bucket}"

sudo -u flink /opt/flink/current/bin/flink stop \
  --savepointPath "s3p://${ICEBERG_BUCKET}/flink/savepoints/aws-iceberg" \
  "${FLINK_JOB_ID}"

再起動方法は運用に使用するsubmit方式とFlink versionに合わせる。復元後はKafkaを初期offsetから読み直していないこと、checkpoint成功、件数、Iceberg snapshotを確認する。

exactly-onceとmaintenanceの注意

IcebergのFlink sinkはcheckpointと連携してexactly-once commitを行う。ただし次の運用は重複や欠落の原因になる。

  • checkpointを使わず、毎回earliest-offsetから新規jobとして起動する
  • savepointを取らずにgroup IDを変更する
  • job稼働中にsnapshotを短時間でexpireする
  • jobが使用中の一時fileをorphan fileとして削除する
  • 同じKafka recordを別jobから同じappend tableへ書く

Flink streaming writeはIceberg snapshot summaryに最後にcommitしたcheckpoint IDを保持する。snapshot expirationではFlink jobが作成した最後のsnapshotを残し、orphan file削除の保持時間を復旧期間より十分長くする。

small fileとGlue更新頻度

Flinkはcheckpoint完了ごとにIcebergへcommitする。間隔を短くすると遅延は減るがsnapshotとsmall fileが増える。

まずは5分間隔で開始し、次を監視する。

  • チェックポイントの所要時間
  • data file数と平均size
  • Glue UpdateTable回数
  • Athena scan量
  • commit失敗とretry

定期的にcompaction、snapshot expiration、orphan file削除を行う。ただしFlink jobの復旧に必要なsnapshotと一時fileを削除しない。

トラブルシュート

Glueへ接続できない

AWS_REGION、credential file、JobManager/TaskManagerのEnvironmentFile、Glue権限、各nodeからGlue endpointへのHTTPS接続を確認する。

S3へ書けない

S3 List/Get/Put/Delete、bucket policy、KMS利用時のKMS権限、全TaskManagerからS3への接続を確認する。

S3にfileはあるがAthenaから見えない

checkpointとIceberg commit前のdata file、または失敗jobの一時fileである可能性がある。S3 objectの存在だけで成功と判断せず、Flink UI、Glue table、Iceberg snapshot、Athenaの順に確認する。

NoSuchMethodErrorが発生する

Flink、Iceberg runtime、AWS bundleのversion不整合や複数versionのJAR混在を確認する。

find /opt/flink/current/lib -maxdepth 1 -type f \
  \( -name '*iceberg*.jar' -o -name '*aws*.jar' \) \
  -printf '%f\n' | sort

まとめ

Kafkaのsyslog/authlogをFlink SQLで読み、AWS Glue Data CatalogとAWS S3上のIcebergへ直接書き込んだ。

重要な点は次である。

  • Flink用Iceberg runtimeとAWS bundleを全nodeへ配置する
  • JobManagerとTaskManagerの両方へAWS credentialとregionを渡す
  • durableなcheckpointを有効にする
  • 複数の無限streamはSTATEMENT SETでsubmitする
  • checkpoint完了後にGlue、S3、Athenaの順で確認する
  • snapshot/orphan cleanupでFlinkの復旧状態を壊さない

日次batch tableとrealtime tableを分離することで、既存のAWS移行処理を止めずにFlinkからの直接書き込みを検証できる。

参考資料

関連記事

実機検証について

本記事の手順は、執筆時点では実機環境での一連の動作検証をまだ完了していない。掲載している設定値、command、IAM権限、JAR構成は設計例を含むため、利用するFlink、Iceberg、Hadoop AWS、AWS SDKなどのversion互換性と実環境の構成に合わせた確認が必要である。

実施する場合は、検証環境でbackupとrollback手順を用意し、checkpoint、savepoint、Iceberg commit、障害復旧を確認してから本番環境へ適用する。実機検証の結果に応じて本記事を更新する予定である。

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?