はじめに

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

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

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

作成する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からの直接書き込みを検証できる。

参考資料

関連記事

【導入編】HDFSログ分析基盤をAWS S3へ段階移行する全8編の構成を紹介する
【第3編】Apache OzoneのローカルS3からAWS S3へrawログとIcebergを移行してみる
【第5編】ログ分析基盤に残るHDFS依存をAWS S3へ移行してみる
syslog/authlogをリアルタイム解析できるようにする
Apache Flink + Iceberg で syslog/authlog の1分集計テーブルを作り、Trino/Grafana を高速化する

実機検証について

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

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