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)

【第5編】ログ分析基盤に残るHDFS依存をAWS S3へ移行してみる

0
Last updated at Posted at 2026-09-19

はじめに

これまでの記事で、次の移行経路を作成した。

また、FlinkからAWS S3上のIcebergへ直接書き込む経路と、TrinoからAWS Glue + S3 Icebergを参照するcatalogも追加した。

しかし、data fileをコピーしただけではHDFSを撤去できない。継続的にHDFSへ書き込むservice、checkpoint、batch、運用scriptが1つでも残っていると、HDFS停止後に処理が失敗する。

本記事では、ログ分析基盤に残るHDFS依存をAWS S3へ移行する。

HDFS serviceの停止とmodule削除は次の記事で扱い、本記事では依存先の切り替えまでを対象とする。

対象となるHDFS依存

今回確認するのは次である。

依存元 従来のHDFS用途 移行先
Kafka Connect raw JSON Text Amazon S3 raw bucket
Spark batch raw入力、Iceberg出力 Amazon S3 raw、Glue Iceberg
Flink Iceberg出力 Glue Iceberg
Flink checkpoint/savepoint Amazon S3
Hive raw/curated/external table Glue Icebergまたは廃止
Trino HDFS Iceberg参照 Glue + Amazon S3
運用script 存在確認、rotation、再投入 AWS CLI、Spark、Iceberg maintenance
backup/restore HDFS間コピー S3 versioning、別bucket、論理restore

移行後の構成

hdfsex-hdfs-to-aws-overview.png

最初にHDFS依存をinventoryする

設定ファイル、script、cron、systemd unitを検索する。

sudo rg -n -i \
  'hdfs://|/data/kafka|/warehouse/iceberg|/apps/flink|hdfs dfs|distcp' \
  /etc /opt /usr/local /var/spool/cron \
  2>/dev/null

systemd timerとcronを確認する。

systemctl list-timers --all
sudo crontab -l
sudo -u spark crontab -l
sudo -u flink crontab -l
sudo -u hadoop crontab -l

HDFS内のtop-level使用量を記録する。

sudo -u hadoop hdfs dfs -du -h /
sudo -u hadoop hdfs dfs -du -h /data
sudo -u hadoop hdfs dfs -du -h /warehouse
sudo -u hadoop hdfs dfs -du -h /apps
sudo -u hadoop hdfs dfs -du -h /user

HDFS上の内容ごとに、移行する、再生成できる、不要、用途不明のいずれかを記録する。用途不明のdirectoryは削除候補にしない。

AWS側の保存先を確認する

export AWS_REGION='ap-northeast-1'
source /etc/rawlog-s3-backup/common.env
export AWS_RAW_BUCKET="${S3_BUCKET_NAME}"
export AWS_RAW_PROFILE=home-rawlog-backup
source /etc/iceberg/aws.env
export AWS_ICEBERG_BUCKET="${ICEBERG_BUCKET}"
export AWS_ICEBERG_PROFILE=home-aws

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

aws s3api head-bucket \
  --bucket "${AWS_RAW_BUCKET}" \
  --profile "${AWS_RAW_PROFILE}"

aws s3api head-bucket \
  --bucket "${AWS_ICEBERG_BUCKET}" \
  --profile "${AWS_ICEBERG_PROFILE}"

aws glue get-database \
  --name "${GLUE_DATABASE}" \
  --region "${AWS_REGION}" \
  --profile "${AWS_ICEBERG_PROFILE}"

既存Raw backupとIcebergを移行先として使用するため、bucketやbatch Iceberg tableは新規作成しない。Kafka Connect、Spark、Flinkにはローカルprofileを配布せず、それぞれの実行主体へ対象prefixだけの権限を付与する。

Kafka Connect HDFS SinkをS3 Sinkへ移行する

いきなりHDFS Sinkを停止しない

移行期間中はHDFS SinkとS3 Sinkを並行稼働させる。

connectorごとにconsumer groupとoffsetが独立するため、同じtopicを2つのsinkへ保存できる。

S3 Sink pluginを導入する

Kafka Connect workerへConfluent S3 Sink Connectorを導入する。使用中のKafka Connect、Java、Confluent Platformと互換性のあるversionを選ぶ。

plugin導入後、workerを再起動してclassを確認する。

curl -s http://connect1:8083/connector-plugins \
  | jq -r '.[].class' \
  | grep 'io.confluent.connect.s3.S3SinkConnector'

distributed workerが複数ある場合は、全workerへ同じplugin versionを配置する。

AWS credentialをConnectへ渡す

connector JSONへaccess keyを直接記載せず、Connect serviceのAWS default credential chainから取得する。bare metal検証環境ではsystemd EnvironmentFileを使用できる。

sudo tee /etc/kafka-connect/aws.env >/dev/null <<'EOF'
AWS_REGION=ap-northeast-1
AWS_DEFAULT_REGION=ap-northeast-1
AWS_ACCESS_KEY_ID=<CONNECT_AWS_ACCESS_KEY_ID>
AWS_SECRET_ACCESS_KEY=<CONNECT_AWS_SECRET_ACCESS_KEY>
EOF

sudo chown root:kafka /etc/kafka-connect/aws.env
sudo chmod 640 /etc/kafka-connect/aws.env

Kafka Connectのsystemd unitへ追加する。

[Service]
EnvironmentFile=/etc/kafka-connect/aws.env

本番では長期access keyではなくIAM Roles Anywhereなどの短期credentialを検討する。

syslog用S3 Sinkを作成する

timestamp.extractor=Recordとrotate.interval.msを使用し、決定的なtime partitionを作る。Wallclockとschedule rotationの組み合わせは再処理時の配置が変わる可能性がある。

sudo tee /tmp/s3-syslog-sink.json >/dev/null <<'EOF'
{
  "name": "s3-syslog-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "3",
    "topics": "syslog",
    "s3.bucket.name": "<YOUR_UNIQUE_RAW_BUCKET>",
    "s3.region": "ap-northeast-1",
    "topics.dir": "raw",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "partition.duration.ms": "3600000",
    "path.format": "YYYY/MM/dd/HH",
    "locale": "ja-JP",
    "timezone": "Asia/Tokyo",
    "timestamp.extractor": "Record",
    "rotate.interval.ms": "3600000",
    "flush.size": "10000",
    "schema.compatibility": "NONE",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}
EOF

curl -sS -X POST \
  -H 'Content-Type: application/json' \
  --data @/tmp/s3-syslog-sink.json \
  http://connect1:8083/connectors

object pathは次のようになる。

s3://<YOUR_UNIQUE_RAW_BUCKET>/raw/syslog/YYYY/MM/DD/HH/...

authlog用S3 Sinkを作成する

syslog用JSONを複製し、次を変更する。

name   : s3-authlog-sink
topics : authlog

curl -sS -X POST \
  -H 'Content-Type: application/json' \
  --data @/tmp/s3-authlog-sink.json \
  http://connect1:8083/connectors

S3 Sinkを確認する

for connector in s3-syslog-sink s3-authlog-sink; do
  curl -s "http://connect1:8083/connectors/${connector}/status" | jq
done

aws s3 ls \
  "s3://${AWS_RAW_BUCKET}/raw/syslog/" \
  --recursive \
  --profile "${AWS_RAW_PROFILE}"

aws s3 ls \
  "s3://${AWS_RAW_BUCKET}/raw/authlog/" \
  --recursive \
  --profile "${AWS_RAW_PROFILE}"

S3上のobjectが確定するまで、flush.sizeまたはrotation条件を満たす必要がある。少量環境では検証用に値を小さくし、確認後に戻す。

Spark batchのraw入力をAWS S3へ変更する

既存の直接ロードではHDFS rawを読んでいた。

- hdfs://cluster1/data/kafka/syslog/${DAY}
+ s3a://${AWS_RAW_BUCKET}/raw/syslog/${DAY}

- hdfs://cluster1/data/kafka/authlog/${DAY}
+ s3a://${AWS_RAW_BUCKET}/raw/authlog/${DAY}

Iceberg出力はAWS Glue catalogへ変更する。

- hive_prod.logs.syslog_iceberg
+ glue_prod.logs.syslog_iceberg

- hive_prod.logs.authlog_iceberg
+ glue_prod.logs.authlog_iceberg

Sparkのtext sourceでS3 rawを読めることを確認する。

sudo -u spark /usr/local/bin/spark-sql-iceberg-aws <<EOF
CREATE OR REPLACE TEMPORARY VIEW syslog_s3_raw
USING text
OPTIONS (
  path 's3a://${AWS_RAW_BUCKET}/raw/syslog/2026/09/15'
);

SELECT COUNT(*) AS cnt
FROM syslog_s3_raw
WHERE value IS NOT NULL
  AND length(trim(value)) > 0;
EOF

既存の日次script、EnvironmentFile、systemd timerで次を変更する。

  • 移行元のパス
  • 移行先のカタログ
  • HDFS存在確認をS3 prefix確認へ変更
  • hdfs dfsによるrotationをS3 lifecycleまたはIceberg maintenanceへ変更
  • HDFSユーザーでの実行をSpark専用IAM principalへ変更

同一日をHDFS版とAWS版で実行し、件数、MIN(ts)、MAX(ts)、host/severity/hr別分布を比較する。

FlinkのHDFS依存を移行する

FlinkからAWS S3 Icebergへ書く新規jobだけでなく、既存jobをすべてinventoryする。

sudo -u flink /opt/flink/current/bin/flink list
sudo rg -n -i \
  'hdfs://|hive_prod|/apps/flink|warehouse/iceberg' \
  /opt/flink/jobs /etc/flink

対象には次が含まれる。

  • syslog event明細job
  • authlog event明細job
  • syslog 1分集計job
  • authlog 1分集計job
  • 検証用に残ったSQL job
  • checkpoint/savepoint設定

新規AWS jobを先に起動する

4th記事の手順で次を作成する。

glue_prod.logs.syslog_events
glue_prod.logs.authlog_events
s3p://<YOUR_ICEBERG_BUCKET>/flink/checkpoints/aws-iceberg
s3p://<YOUR_ICEBERG_BUCKET>/flink/savepoints/aws-iceberg

既存HDFS jobとAWS jobを並行稼働し、同じ時間帯の件数を比較する。

1分集計tableもAWSへ移す

必要であればAWS側に次を作成する。

glue_prod.logs.syslog_host_1m
glue_prod.logs.authlog_host_1m

既存jobのcatalogをhive_prodからglue_prodへ変更し、checkpoint pathもS3へ変更する。event tableだけを移して1分集計jobを忘れない。

savepointを取得して旧jobを停止する

export HDFS_JOB_ID="${HDFS_JOB_ID:?set the running HDFS-backed 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/hdfs-final" \
  "${HDFS_JOB_ID}"

停止後もAWS jobのcheckpoint、Kafka lag、Iceberg commitが正常であることを確認する。旧HDFS checkpoint directoryはrollback期間が終わるまで削除しない。

Hive依存を分類する

Hive Metastore内のdatabaseとtableを一覧化する。

SHOW DATABASES;
SHOW TABLES IN logs;
SHOW TABLES IN authlogs;
DESCRIBE FORMATTED logs.syslog_curated;
DESCRIBE FORMATTED authlogs.authlog_curated;

tableを次のように分類する。

種類 対応
廃止済みraw/parsed/curated 参照停止後にDROP
HDFS Iceberg Glue Icebergへ論理移行
HDFS Parquetで継続利用 Glue IcebergへCTASまたはSparkで変換
一時・検証table 所有者確認後に削除
用途不明 削除せず調査

Hive MetastoreはOzone catalogで使っている間は停止できない。すべてAWS Glueへ移行し、Trino、Spark、Flinkの参照がなくなった後に撤去候補となる。

Trino、Grafana、Zeppelinを切り替える

3rd記事で追加したiceberg_aws catalogへ参照を変更する。

- FROM iceberg.logs.syslog_iceberg
- FROM iceberg_ozone.logs_s3.syslog_iceberg
+ FROM iceberg_aws.logs.syslog_iceberg

realtime tableは次を参照する。

SELECT dt, hr, COUNT(*) AS cnt
FROM iceberg_aws.logs.syslog_events
GROUP BY dt, hr
ORDER BY dt DESC, hr DESC;

Zeppelinでは第3編で作成したtrino_aws JDBC Interpreterを使用する。接続先は従来と同じTrino Coordinatorで、JDBC URLのdefault catalogとschemaをAWS側へ変更する。

default.driver=io.trino.jdbc.TrinoDriver
default.url=jdbc:trino://trino1:8080/iceberg_aws/logs
default.user=zeppelin

%trino_aws
SELECT current_catalog, current_schema;

SELECT COUNT(*)
FROM syslog_iceberg
WHERE dt = current_date - INTERVAL '1' DAY;

通常のJDBC通信ではZeppelinはTrino経由で参照するため、Zeppelin hostへIceberg data bucket用のAWS credentialやS3 client設定を追加しない。Glue/S3への接続とIAM権限はTrino全nodeに設定する。Trinoのspooling protocolを有効にしている場合だけ、Zeppelinからspooling用storageへの接続を別途設定する。

Grafana dashboardとZeppelin Notebookは複製して比較し、HDFS/Ozone版をすぐに削除しない。

運用scriptをAWSへ変更する

次を1つずつ置き換える。

HDFS処理 AWS移行後
hdfs dfs -test -e aws s3api list-objects-v2またはSpark count
hdfs dfs -count aws s3 ls --recursive --summarize
hdfs dfs -rm -r S3 lifecycleまたは対象prefixを確認した削除
HDFS raw rotation S3 lifecycle rule
HDFS Iceberg file削除 Iceberg maintenance procedure
HDFS間DistCp S3 replication、rclone、AWS DataSyncなど
HDFS再投入 S3 rawからSparkで対象日を再作成

Iceberg tableのdata/metadata directoryをaws s3 rm --recursiveで直接rotationしない。DELETE、snapshot expiration、orphan file削除をIceberg経由で行う。

具体的なS3 Lifecycle、Icebergのdata保持、compaction、snapshot expiration、orphan file削除は、独立記事「Amazon S3のrawログとApache Icebergに保持期間とメンテナンスを設定する」で設定する。Rawは7日、batch Icebergは14日、Flink realtime Icebergは当日分を保持する。

HDFS Sinkを停止する

すべての比較完了後、既存HDFS Sinkを停止する。

curl -sS -X PUT \
  http://connect1:8083/connectors/hdfs-syslog-sink/pause

curl -sS -X PUT \
  http://connect1:8083/connectors/hdfs-authlog-sink/pause

実際のconnector名へ置き換える。pause後に確認する。

curl -s http://connect1:8083/connectors/hdfs-syslog-sink/status | jq
curl -s http://connect1:8083/connectors/hdfs-authlog-sink/status | jq

一定期間問題がなければconnectorを削除する。

curl -sS -X DELETE \
  http://connect1:8083/connectors/hdfs-syslog-sink

curl -sS -X DELETE \
  http://connect1:8083/connectors/hdfs-authlog-sink

connector削除はHDFS上の既存objectを削除しない。

移行完了判定

HDFS停止記事へ進む条件は次である。

  • Kafka Connect HDFS Sinkが停止または削除済み
  • S3 Sinkがsyslog/authlogを継続保存している
  • Spark batchがS3 rawだけを読んでいる
  • Sparkの出力がGlue Icebergだけになっている
  • 全Flink jobのsinkとcheckpoint/savepointがS3になっている
  • 1分集計jobも移行済みまたは廃止済み
  • Trino/Grafanaがiceberg_awsを参照している
  • Zeppelinのtrino_aws Interpreterからbatch/realtime Iceberg tableを参照できる
  • Hive tableを移行、廃止、保留に分類済み
  • cron、timer、shellからhdfs://とhdfs dfsがなくなっている
  • 再投入とrestoreをS3から実行できる
  • 一定期間HDFSへの新規書き込みがない

まとめ

HDFS撤去前に必要なのはdata copyだけではなく、継続的なwriterと運用処理の移行である。

すべての依存を切り替え、HDFSへの新規accessがないことを確認してから、停止とmodule削除へ進む。

参考資料

関連記事

実機検証について

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

実施する場合は、HDFS側とAWS側を並行稼働できる検証環境で、rawログ、Iceberg table、checkpoint、query結果を比較してから旧処理を停止する。実機検証の結果に応じて本記事を更新する予定である。

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?