はじめに

これまで、自宅のログ収集・分析基盤では、syslogとauth.logをKafkaへ送り、HDFSへ保存したrawログをSparkで変換してApache Icebergへ格納してきた。

今回、この基盤をHDFS中心の構成からAmazon S3中心の構成へ段階的に移行することにした。

ただし、HDFSからS3へdata fileをコピーするだけでは移行は完了しない。実際の基盤には、次のような依存がある。

Kafka Connectによるrawログの継続保存
Sparkの日次batchとHive中間テーブル
Icebergのdata fileとmetadata
Flinkのrealtime table、checkpoint、savepoint
Hive MetastoreまたはAWS Glue Data Catalog
Trino、Grafana、Zeppelinからの参照
rotation、再投入、backup、監視などの運用処理

これらを一度に切り替えると、障害時に原因と影響範囲を切り分けにくい。そのため、構造の単純化、保存先の移行、処理の切り替え、旧基盤の撤去、検証用AWS resourceのcleanupを7編に分けた。

本記事では、各記事の役割、実施順序、最終的な構成を紹介する。

移行前の構成

移行前は、Kafka ConnectがHDFSへrawログを書き込み、Hiveの中間テーブルを経由してSparkがIcebergテーブルを作成していた。

この構成には、次の課題があった。

raw、Hive中間テーブル、IcebergがHDFSへ集中している
中間テーブルの生成と削除が運用負荷になる
HDFSを止めるとbatch、stream、queryのすべてへ影響する
AWS上のAthenaやGlueと直接連携しにくい
HDFS廃止の対象範囲を判断しにくい

移行後の構成

最終的には、rawログとIcebergをAmazon S3へ格納し、IcebergのcatalogにはAWS Glue Data Catalogを利用する。

HDFSは移行期間中の比較・rollback先として残し、すべての依存をAWS側へ切り替えた後に停止・撤去する。

全8編の構成

第1編: Hive中間テーブルを廃止する

「Hiveの中間テーブルを廃止してHDFSからApache Icebergへ直接取り込んでみる」では、HDFS上のraw JSON TextをSpark SQLで直接読み、Icebergへ書き込む。

既存Hive Viewに含まれていたJSON抽出、timestamp整形、severity判定、partition列生成をSpark側へ移植する。旧方式と新方式の件数、時刻範囲、集計結果を比較してから中間テーブルを廃止する。

最初にこの整理を行うことで、後続記事では主にrawの入力pathとIcebergのcatalog・warehouseを変更すればよくなる。

第2編: HDFSからローカルS3へ移行する

「HDFSのrawログとApache IcebergをローカルS3へ移行してみる」では、Apache OzoneのS3 GatewayをローカルS3互換storageとして利用する。

HDFS rawをOzoneへコピーする
SparkからS3A経由でrawを読む
Ozone側に新しいIceberg catalogを作る
IcebergをSpark SQLで論理コピーする
TrinoへOzone用catalogを追加する
GrafanaとZeppelinを新catalogへ切り替える

MinIOは使用せず、Hadoop ecosystemとの連携を検証しやすいApache Ozoneを採用している。

この段階ではHDFSを正本として残し、S3 API、credential、path-style access、Iceberg metadataの扱いをローカル環境で確認する。

第3編: ローカルS3からAWS S3へ移行する

「Apache OzoneのローカルS3からAWS S3へrawログとIcebergを移行してみる」では、Ozoneで検証した構成をAWSへ移す。

rawログをrcloneでAmazon S3へコピーする
object数、合計size、sample checksumを比較する
AWS Glue Data CatalogをSparkへ追加する
AWS側でIceberg tableとmetadataを新しく生成する
TrinoへGlueとnative S3を使うcatalogを追加する
GrafanaとZeppelinをAWS側へ切り替える

Iceberg warehouseをファイル単位でそのままコピーせず、Spark SQLでAWS側へ論理コピーする点が重要になる。

第4編: FlinkからAWS S3 Icebergへ書き込む

「Apache FlinkからAWS S3上のApache Icebergへリアルタイムに書き込んでみる」では、Kafka上のsyslogとauthlogをFlink SQLで処理し、AWS S3上のIcebergへ直接書き込む。

Flink用Iceberg runtimeとAWS bundleを配置する
JobManagerとTaskManagerへAWS認証情報を渡す
checkpointとsavepointをAmazon S3へ保存する
AWS Glueをcatalogとして利用する
AthenaとTrinoからcommit結果を確認する
savepointを使った停止・再開を確認する

日次batch tableとrealtime tableを分離し、既存処理を維持しながら検証する。

第5編: 残っているHDFS依存をAWSへ移行する

「ログ分析基盤に残るHDFS依存をAWS S3へ移行してみる」では、data移行後も残っているwriter、reader、運用処理をAWS側へ切り替える。

ここで初めて、HDFSへの新しい読み書きが発生しない状態を目指す。

第6編: HDFSを停止・撤去する

「AWS S3移行後にHDFSを安全に停止・撤去してみる」では、HDFSを停止した状態で通常運用できることを確認し、不要になったmoduleを段階的に撤去する。

撤去は次の4段階に分ける。

HDFS向け設定とjobを無効化する
HDFS daemonを停止し、データと設定を保持して観察する
HDFS server packageを削除する
rollback期限後にHDFS data directoryを削除する

NameNode、DataNode、JournalNode、ZKFCなどのHDFS専用moduleは条件を満たせば削除できる。一方、hadoop-aws、YARN、Hive Metastore、ZooKeeperは他の処理が利用している可能性があるため、HDFSと一括して削除しない。

第7編: S3とIcebergの保持・メンテナンスを設定する

「Amazon S3のrawログとApache Icebergに保持期間とメンテナンスを設定する」では、S3移行後にdataが増え続けないよう、保持期間と定期maintenanceを設定する。

S3 rawはLifecycleで7日保持する
batch Icebergは当日を含む14日保持する
Flink realtime Icebergは当日分を保持する
compaction、snapshot expiration、orphan file削除を実行する
Flinkのcheckpoint復旧に必要なsnapshotと一時fileを保護する

第8編: 検証用AWS環境を削除する

「検証で作成したAWS S3・Glue・IAM環境を安全に削除してみる」では、seriesの実機検証だけに使用したAWS resourceを、依存関係の逆順で撤去する。

AWSへ書き込むKafka Connect、Spark、Flinkを停止する
Glue TableとDatabaseを削除する
S3のmultipart upload、全object version、delete markerを削除する
空になったS3 bucketを削除する
検証専用IAM credential、user、role、policyを削除する
必要に応じて検証専用KMS keyの削除をscheduleする
local hostのAWS設定を撤去し、料金を確認する

第8編は検証環境を残さない場合だけ実施する。本番移行先としてAWSを利用し続ける場合は実施しない。

後続のAWSマネージドサービス移行との関係

本シリーズの第1〜7編を完了すると、rawログ、Iceberg、checkpoint、catalog、query、運用処理がAmazon S3とAWS Glue Data Catalogへ切り替わり、保持・maintenanceも設定された状態になる。この状態を、後続の「オンプレのログ分析基盤をAWSマネージドサービスへ移行する」シリーズの開始地点とする。

後続シリーズへ進む場合も、本シリーズ第1〜7編の構築手順は順番に実施する。第5編のKafka Connect、Spark、Flink、query基盤の切り替えも省略しない。後続シリーズでは、ここで構築したS3/Glue data基盤を作り直さず、オンプレで稼働しているKafka、Kafka Connect、Spark、Flink、TrinoをAWSマネージドサービスへ置き換える。

後続シリーズへ進む場合にスキップする手順

第8編のcleanupだけは実施しない。第8編を実施すると、後続シリーズが引き継ぐS3 bucket、Glue Database/Table、IAM resourceが削除されるためである。

どの順番で読むか

既存のS3バックアップを使用する場合（本seriesの推奨経路）

すでに次の記事でHDFS rawとIcebergをAWSへバックアップしている場合は、第2編のOzone構築と第3編の再コピーを実施しない。

Raw確認にはhome-rawlog-backup、Icebergの同期・参照にはhome-aws、AWS resourceの管理操作にはhome-aws-admin profileを使う。第8編のS3/Glue削除は、既存バックアップを今後も利用するため実施しない。

ローカルS3を経由して段階的に検証する場合

移行、HDFS撤去、maintenance設定は第1編から第7編まで順番に実施する。検証用AWS環境も削除する場合だけ、最後に第8編を実施する。

オンプレミス内でS3互換性を先に確認でき、AWS移行時の変更点を分離しやすい。AWSへ常時接続できない検証環境にも向いている。

HDFSからAWSへ直接移行する場合

Apache Ozoneによるローカル検証が不要なら、第2編を構築手順として実施しない。既存S3 backupがある場合は、第3編のAWS側準備と論理移行も再実行せず、bucket、Glue table、件数、時刻範囲の確認だけを行う。

ただし、第3編のrawコピー例はOzoneをsourceとしている。そのため、HDFSから直接コピーする場合は、既存のDistCp、AWS DataSync、またはSparkによる読み書きへ置き換える必要がある。Icebergはこの場合も、metadata directoryの単純コピーではなくAWS側tableへの論理コピーを基本とする。

Flinkを利用していない場合

第4編のrealtime table構築は省略できる。ただし、第5編と第6編では、Flink以外も含めてHDFSを参照するjob、checkpoint、scriptが残っていないことを確認する。

このseriesで共通して守ること

旧環境をすぐに削除しない

新旧を一定期間並行稼働させ、同じ対象期間について比較する。確認項目は、data形式に応じて次を組み合わせる。

record件数
MIN(timestamp)とMAX(timestamp)
日付、時刻、host、severity別の集計
object数と合計size
sample fileのchecksum
Iceberg snapshotとmetadata location
Kafkaコンシューマーの遅延
Flink checkpointの成功時刻

rawとIcebergを同じ方法でコピーしない

rawログはimmutableなobjectとして物理コピーできる。一方、Icebergはdata fileだけでなくmanifest、metadata、catalog上のpointerを持つ。

そのため、本seriesでは次の方針を採る。

rawログ: object copy + 件数・size・checksum確認
Iceberg : 新catalogにtableを作成 + SQLによる論理コピー

credentialを設定ファイルへ直接埋め込まない

記事中では検証環境向けのprofileやsystemd EnvironmentFileも扱うが、本番環境ではEC2 instance profile、IAM Roles Anywhereなどの短期credentialを優先する。

停止と削除を同時に行わない

HDFSへのaccessがゼロになっても、すぐにdata directoryを削除しない。停止状態で通常のbatch周期、週次・月次処理、障害復旧を確認してから、packageとdataを別工程で撤去する。

完了条件

series全体の完了条件は次のとおりである。

Kafka ConnectがAmazon S3へrawログを継続保存している
SparkがAmazon S3のrawを読み、AWS側Icebergへ書き込める
Flink利用時はS3 checkpointから復旧できる
IcebergのcatalogがAWS Glueへ切り替わっている
Trino、Athena、Grafana、ZeppelinからAWS側を参照できる
HDFSを停止しても通常処理と定期処理が失敗しない
HDFS専用moduleと共有moduleの削除範囲を区別できている
rollback期限終了後にHDFS data directoryを撤去できる

まとめ

このseriesでは、HDFSからAmazon S3への移行を単なるdata copyとして扱わず、table設計、writer、reader、metadata、checkpoint、query、運用、旧module撤去までを1つのmigrationとして扱う。

最初にHive中間テーブルを廃止してdata flowを単純化し、ローカルS3とAWS S3で段階的に検証する。その後、Flinkと残存HDFS依存をAWSへ切り替え、最後にHDFSを停止・撤去する。

各段階で旧環境を残して比較するため、問題が発生しても直前の構成へ戻しやすい。すべてを一度に変更するのではなく、data、処理、参照、運用、撤去を分けることが、このseriesの基本方針である。

関連記事

実機検証について

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

実施する場合は、検証環境でbackupとrollback手順を用意し、各編の比較項目を確認してから次の段階へ進む。本番環境へ適用する前に実機検証を行い、結果に応じて本seriesの内容を更新する予定である。