【第1編】Hiveの中間テーブルを廃止してHDFSからApache Icebergへ直接取り込んでみる

はじめに

本seriesでは、後続のAWS移行時に新しいbucketへ履歴をコピーし直さず、既存のS3 Raw backupとS3 Icebergを使用する。Rawはhome-rawlog-backup、Icebergはhome-aws profileで確認する。本稿の変更はHDFS側の変換処理を整理するもので、既存AWS backupを上書きしない。

これまで構築したログ収集・分析基盤では、Kafka Connect HDFS Sinkで保存したrawログをHiveで解析し、一度Parquetのcuratedテーブルへ変換してからApache Icebergへ投入していた。

今回は次のHive中間テーブルを廃止し、Spark SQLでHDFS rawを直接解析して既存のIcebergテーブルへ投入する。

• logs.syslog_curated
• authlogs.authlog_curated

syslogだけでなくauthlogも対象とする。今回はS3化、各ミドルウェアの構築、Icebergテーブルの初期構築は行わない。

前提

以下は構築済みとする。

• Kafka Connect HDFS Sinkからhdfs://cluster1/data/kafka/syslogとhdfs://cluster1/data/kafka/authlogへrawログを保存できる
• /usr/local/bin/spark-sql-icebergを使用できる
• Iceberg catalog hive_prodを使用できる
• hive_prod.logs.syslog_icebergとhive_prod.logs.authlog_icebergを作成済み
• 操作ホストはope1

既存のIcebergテーブルはどちらも次の列を持つ。

host string, ts TIMESTAMP_NTZ, severity int, program string,
msg string, dt date, hr int

現在の構成

変更前は次の流れである。

syslogでは次を使用している。

authlogでは次を使用している。

今回変更する理由

既存方式では、同じ日次データをHive curatedとIcebergの両方に保持する。また、rawテーブルのLOCATION変更、parsed view、curated生成SQL、curatedローテーション、curatedからIcebergへの投入を順番に運用する必要がある。

Icebergが長期履歴分析層として稼働しているため、今回は解析処理をSpark SQLへ移し、HDFS rawからIcebergまでを1回のジョブで処理する。

変更後の構成

Hive Metastoreはhive_prod catalogで引き続き使用する。Hive curatedを廃止しても、Hive/Hive Metastore自体を撤去する構成ではない。

既存Hive curated処理の確認

HDFS rawの形式

syslog用Connectorとauthlog用Connectorは、いずれも次の設定で保存している。

• format.class: io.confluent.connect.hdfs.json.JsonFormat
• partitioner.class: io.confluent.connect.storage.partitioner.TimeBasedPartitioner
• path.format: yyyy/MM/dd/HH
• timezone: Asia/Tokyo
• timestamp.extractor: Wallclock
• topics.dir: /data/kafka

したがって対象日の入力パスは次になる。

hdfs://cluster1/data/kafka/syslog/YYYY/MM/DD
hdfs://cluster1/data/kafka/authlog/YYYY/MM/DD

Hiveではline STRINGのTEXTFILEとして読み、JSONから次の4項目を抽出している。

{
  "ts": "2026-07-11T01:05:14+09:00",
  "src_host": "raspi4-3",
  "program": "tubu-frontend",
  "message": "..."
}

対応は次の通りである。

raw JSON	Iceberg
src_host	host
ts	ts、dt、hr
program	program
message	msg、severity

parsed viewで行っていた処理

logs.v_syslog_parsed_dayとauthlogs.v_authlog_parsed_dayは同じ変換を行っている。

• 空行を除外する
• src_host、ts、program、messageをget_json_objectで抽出する
• tsの先頭19文字のTを空白へ置換する
• ts_rawの先頭10文字からdtを作る
• message中のキーワードからseverityを0～7へ数値化する
• host、timestamp、messageが不正な行を除外する

severityの判定順とデフォルト値は既存Viewを変更せず移植する。

timestampの扱い

rawのtsは2026-07-11T01:05:14+09:00形式である。既存処理では次の役割に分けていた。

• ts_raw: raw JSONの元文字列
• ts_text: yyyy-MM-dd HH:mm:ss形式のローカル時刻文字列
• ts: Hive内の確認用TIMESTAMP

Iceberg投入時はHive ParquetのTIMESTAMPを渡さず、ts_text相当の文字列をSparkでTIMESTAMP_NTZへ変換している。今回もこの方針を維持する。

CAST(
  regexp_replace(substr(ts_raw, 1, 19), 'T', ' ')
  AS TIMESTAMP_NTZ
)

Spark session timezoneは既存設定と同じAsia/Tokyoにする。+9 hour、-9 hour、to_utc_timestamp、from_utc_timestampは追加しない。

syslogをHDFS rawから直接変換する

以下は1日分を削除後、再投入できる完全なSpark SQLである。load.dtとload.dayを対象日に変更して実行する。

sudo -u spark /usr/local/bin/spark-sql-iceberg <<'EOF'
SET spark.sql.session.timeZone=Asia/Tokyo;
SET load.dt=2026-09-15;
SET load.day=2026/09/15;

CREATE OR REPLACE TEMPORARY VIEW syslog_raw_text
USING text
OPTIONS (
  path 'hdfs://cluster1/data/kafka/syslog/${load.day}'
);

CREATE OR REPLACE TEMPORARY VIEW syslog_parsed AS
WITH parsed AS (
  SELECT
    get_json_object(value, '$.src_host') AS host,
    get_json_object(value, '$.ts')       AS ts_raw,
    get_json_object(value, '$.program')  AS program,
    get_json_object(value, '$.message')  AS msg
  FROM syslog_raw_text
  WHERE value IS NOT NULL
    AND length(trim(value)) > 0
), typed AS (
  SELECT
    host,
    CAST(
      regexp_replace(substr(ts_raw, 1, 19), 'T', ' ')
      AS TIMESTAMP_NTZ
    ) AS ts,
    CASE
      WHEN msg RLIKE '(?i)\\b(emerg|emergency)\\b' THEN 0
      WHEN msg RLIKE '(?i)\\balert\\b' THEN 1
      WHEN msg RLIKE '(?i)\\bcrit(ical)?\\b' THEN 2
      WHEN msg RLIKE '(?i)\\b(err|error|failed|failure)\\b' THEN 3
      WHEN msg RLIKE '(?i)\\bwarn(ing)?\\b' THEN 4
      WHEN msg RLIKE '(?i)\\bnotice\\b' THEN 5
      WHEN msg RLIKE '(?i)\\b(info|started|stopped)\\b' THEN 6
      WHEN msg RLIKE '(?i)\\bdebug\\b' THEN 7
      ELSE 6
    END AS severity,
    program,
    msg,
    substr(ts_raw, 1, 10) AS dt_text
  FROM parsed
  WHERE host IS NOT NULL
    AND ts_raw IS NOT NULL
    AND length(ts_raw) >= 19
    AND msg IS NOT NULL
)
SELECT host, ts, severity, program, msg, CAST(dt_text AS DATE) AS dt, HOUR(ts) AS hr
FROM typed
WHERE dt_text = '${load.dt}'
  AND ts IS NOT NULL;

DELETE FROM hive_prod.logs.syslog_iceberg
WHERE dt = DATE '${load.dt}';

INSERT INTO hive_prod.logs.syslog_iceberg
SELECT host, ts, severity, program, msg, dt, hr
FROM syslog_parsed;
EOF

authlogをHDFS rawから直接変換する

authlog側も既存のauthlogs.v_authlog_parsed_dayと同じ処理をSpark SQLへ移す。

sudo -u spark /usr/local/bin/spark-sql-iceberg <<'EOF'
SET spark.sql.session.timeZone=Asia/Tokyo;
SET load.dt=2026-09-15;
SET load.day=2026/09/15;

CREATE OR REPLACE TEMPORARY VIEW authlog_raw_text
USING text
OPTIONS (
  path 'hdfs://cluster1/data/kafka/authlog/${load.day}'
);

CREATE OR REPLACE TEMPORARY VIEW authlog_parsed AS
WITH parsed AS (
  SELECT
    get_json_object(value, '$.src_host') AS host,
    get_json_object(value, '$.ts')       AS ts_raw,
    get_json_object(value, '$.program')  AS program,
    get_json_object(value, '$.message')  AS msg
  FROM authlog_raw_text
  WHERE value IS NOT NULL
    AND length(trim(value)) > 0
), typed AS (
  SELECT
    host,
    CAST(
      regexp_replace(substr(ts_raw, 1, 19), 'T', ' ')
      AS TIMESTAMP_NTZ
    ) AS ts,
    CASE
      WHEN msg RLIKE '(?i)\\b(emerg|emergency)\\b' THEN 0
      WHEN msg RLIKE '(?i)\\balert\\b' THEN 1
      WHEN msg RLIKE '(?i)\\bcrit(ical)?\\b' THEN 2
      WHEN msg RLIKE '(?i)\\b(err|error|failed|failure)\\b' THEN 3
      WHEN msg RLIKE '(?i)\\bwarn(ing)?\\b' THEN 4
      WHEN msg RLIKE '(?i)\\bnotice\\b' THEN 5
      WHEN msg RLIKE '(?i)\\b(info|started|stopped)\\b' THEN 6
      WHEN msg RLIKE '(?i)\\bdebug\\b' THEN 7
      ELSE 6
    END AS severity,
    program,
    msg,
    substr(ts_raw, 1, 10) AS dt_text
  FROM parsed
  WHERE host IS NOT NULL
    AND ts_raw IS NOT NULL
    AND length(ts_raw) >= 19
    AND msg IS NOT NULL
)
SELECT host, ts, severity, program, msg, CAST(dt_text AS DATE) AS dt, HOUR(ts) AS hr
FROM typed
WHERE dt_text = '${load.dt}'
  AND ts IS NOT NULL;

DELETE FROM hive_prod.logs.authlog_iceberg
WHERE dt = DATE '${load.dt}';

INSERT INTO hive_prod.logs.authlog_iceberg
SELECT host, ts, severity, program, msg, dt, hr
FROM authlog_parsed;
EOF

rawのパスを対象日へ限定した上で、JSON内のdtも対象日と一致する行だけを投入している。これは従来のraw_dayのLOCATION変更とparsed viewのWHERE dt='${DT}'を同時に再現するためである。

日次実行スクリプト

既存の/opt/iceberg/bin/load_curated_to_iceberg.shを置き換える日次スクリプトを作る。引数なしではJSTの前日、引数ありでは指定したYYYY-MM-DDを処理する。

本スクリプト自体はope1の一般ユーザーのcrontabから実行し、Spark SQLは既存運用と同じsudo -u spark、rawパスの存在確認はsudo -u hadoopで実行する。実行ユーザーには両方のsudo権限が必要である。入力パスがない場合は、対象日のIcebergデータをDELETEする前に停止する。

sudo tee /opt/iceberg/bin/load_raw_to_iceberg.sh >/dev/null <<'EOF'
#!/usr/bin/env bash
set -euo pipefail

export TZ="${TZ:-Asia/Tokyo}"

DT="${1:-$(date -d 'yesterday' +%F)}"
SPARK_SQL_BIN="${SPARK_SQL_BIN:-/usr/local/bin/spark-sql-iceberg}"

if [[ ! "${DT}" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]] || \
   [[ "$(date -d "${DT}" +%F 2>/dev/null || true)" != "${DT}" ]]; then
  echo "[ERROR] invalid date: ${DT} (expected YYYY-MM-DD)" >&2
  exit 2
fi

DAY="${DT//-//}"
SYSLOG_PATH="hdfs://cluster1/data/kafka/syslog/${DAY}"
AUTHLOG_PATH="hdfs://cluster1/data/kafka/authlog/${DAY}"

log() {
  echo "[INFO] $(date '+%F %T %Z') $*"
}

run_spark() {
  sudo -u spark "${SPARK_SQL_BIN}"
}

for raw_path in "${SYSLOG_PATH}" "${AUTHLOG_PATH}"; do
  if ! sudo -u hadoop hdfs dfs -test -e "${raw_path}"; then
    echo "[ERROR] raw path not found: ${raw_path}" >&2
    exit 3
  fi
done

log "raw to Iceberg start dt=${DT}"

run_spark <<SQL
SET spark.sql.session.timeZone=Asia/Tokyo;

CREATE OR REPLACE TEMPORARY VIEW syslog_raw_text
USING text OPTIONS (path '${SYSLOG_PATH}');

CREATE OR REPLACE TEMPORARY VIEW syslog_parsed AS
WITH parsed AS (
  SELECT
    get_json_object(value, '$.src_host') AS host,
    get_json_object(value, '$.ts')       AS ts_raw,
    get_json_object(value, '$.program')  AS program,
    get_json_object(value, '$.message')  AS msg
  FROM syslog_raw_text
  WHERE value IS NOT NULL AND length(trim(value)) > 0
), typed AS (
  SELECT
    host,
    CAST(regexp_replace(substr(ts_raw, 1, 19), 'T', ' ') AS TIMESTAMP_NTZ) AS ts,
    CASE
      WHEN msg RLIKE '(?i)\\\\b(emerg|emergency)\\\\b' THEN 0
      WHEN msg RLIKE '(?i)\\\\balert\\\\b' THEN 1
      WHEN msg RLIKE '(?i)\\\\bcrit(ical)?\\\\b' THEN 2
      WHEN msg RLIKE '(?i)\\\\b(err|error|failed|failure)\\\\b' THEN 3
      WHEN msg RLIKE '(?i)\\\\bwarn(ing)?\\\\b' THEN 4
      WHEN msg RLIKE '(?i)\\\\bnotice\\\\b' THEN 5
      WHEN msg RLIKE '(?i)\\\\b(info|started|stopped)\\\\b' THEN 6
      WHEN msg RLIKE '(?i)\\\\bdebug\\\\b' THEN 7
      ELSE 6
    END AS severity,
    program,
    msg,
    substr(ts_raw, 1, 10) AS dt_text
  FROM parsed
  WHERE host IS NOT NULL
    AND ts_raw IS NOT NULL
    AND length(ts_raw) >= 19
    AND msg IS NOT NULL
)
SELECT host, ts, severity, program, msg, CAST(dt_text AS DATE) AS dt, HOUR(ts) AS hr
FROM typed
WHERE dt_text = '${DT}' AND ts IS NOT NULL;

DELETE FROM hive_prod.logs.syslog_iceberg WHERE dt = DATE '${DT}';

INSERT INTO hive_prod.logs.syslog_iceberg
SELECT host, ts, severity, program, msg, dt, hr FROM syslog_parsed;
SQL

log "syslog load done dt=${DT}"

run_spark <<SQL
SET spark.sql.session.timeZone=Asia/Tokyo;

CREATE OR REPLACE TEMPORARY VIEW authlog_raw_text
USING text OPTIONS (path '${AUTHLOG_PATH}');

CREATE OR REPLACE TEMPORARY VIEW authlog_parsed AS
WITH parsed AS (
  SELECT
    get_json_object(value, '$.src_host') AS host,
    get_json_object(value, '$.ts')       AS ts_raw,
    get_json_object(value, '$.program')  AS program,
    get_json_object(value, '$.message')  AS msg
  FROM authlog_raw_text
  WHERE value IS NOT NULL AND length(trim(value)) > 0
), typed AS (
  SELECT
    host,
    CAST(regexp_replace(substr(ts_raw, 1, 19), 'T', ' ') AS TIMESTAMP_NTZ) AS ts,
    CASE
      WHEN msg RLIKE '(?i)\\\\b(emerg|emergency)\\\\b' THEN 0
      WHEN msg RLIKE '(?i)\\\\balert\\\\b' THEN 1
      WHEN msg RLIKE '(?i)\\\\bcrit(ical)?\\\\b' THEN 2
      WHEN msg RLIKE '(?i)\\\\b(err|error|failed|failure)\\\\b' THEN 3
      WHEN msg RLIKE '(?i)\\\\bwarn(ing)?\\\\b' THEN 4
      WHEN msg RLIKE '(?i)\\\\bnotice\\\\b' THEN 5
      WHEN msg RLIKE '(?i)\\\\b(info|started|stopped)\\\\b' THEN 6
      WHEN msg RLIKE '(?i)\\\\bdebug\\\\b' THEN 7
      ELSE 6
    END AS severity,
    program,
    msg,
    substr(ts_raw, 1, 10) AS dt_text
  FROM parsed
  WHERE host IS NOT NULL
    AND ts_raw IS NOT NULL
    AND length(ts_raw) >= 19
    AND msg IS NOT NULL
)
SELECT host, ts, severity, program, msg, CAST(dt_text AS DATE) AS dt, HOUR(ts) AS hr
FROM typed
WHERE dt_text = '${DT}' AND ts IS NOT NULL;

DELETE FROM hive_prod.logs.authlog_iceberg WHERE dt = DATE '${DT}';

INSERT INTO hive_prod.logs.authlog_iceberg
SELECT host, ts, severity, program, msg, dt, hr FROM authlog_parsed;
SQL

log "authlog load done dt=${DT}"
log "raw to Iceberg done dt=${DT}"
EOF

sudo chmod 755 /opt/iceberg/bin/load_raw_to_iceberg.sh

任意日を処理する。

/opt/iceberg/bin/load_raw_to_iceberg.sh 2026-09-15

前日分を処理する。

/opt/iceberg/bin/load_raw_to_iceberg.sh

既存の日次ロードは毎日5:00であったため、移行完了後は同じ時刻のエントリを次へ置き換える。

TZ=Asia/Tokyo
0 5 * * * /opt/iceberg/bin/load_raw_to_iceberg.sh >> /var/log/iceberg_load.log 2>&1

rawの保持期間は既存運用で7日である。再投入が必要な日はraw削除前に処理する。

旧方式とのデータ比較

移行検証中はsyslog_curatedとauthlog_curatedを削除しない。まず旧方式で対象日のcuratedを作成し、既存のcurated経由ロード結果と直接ロード結果を比較する。

1. 旧方式の基準値を取得する

sudo -u spark /usr/local/bin/spark-sql-iceberg <<'EOF'
SET spark.sql.hive.convertMetastoreParquet=false;
SET spark.sql.session.timeZone=Asia/Tokyo;

SELECT
  'old_syslog_curated' AS src,
  COUNT(*) AS cnt,
  MIN(CAST(ts_text AS TIMESTAMP_NTZ)) AS min_ts,
  MAX(CAST(ts_text AS TIMESTAMP_NTZ)) AS max_ts
FROM logs.syslog_curated
WHERE dt = '2026-09-15';

SELECT
  'old_authlog_curated' AS src,
  COUNT(*) AS cnt,
  MIN(CAST(ts_text AS TIMESTAMP_NTZ)) AS min_ts,
  MAX(CAST(ts_text AS TIMESTAMP_NTZ)) AS max_ts
FROM authlogs.authlog_curated
WHERE dt = '2026-09-15';
EOF

2. 直接ロードを実行する

/opt/iceberg/bin/load_raw_to_iceberg.sh 2026-09-15

3. Icebergと比較する

sudo -u spark /usr/local/bin/spark-sql-iceberg <<'EOF'
SET spark.sql.hive.convertMetastoreParquet=false;
SET spark.sql.session.timeZone=Asia/Tokyo;

SELECT src, cnt, min_ts, max_ts
FROM (
  SELECT
    'old_syslog_curated' AS src,
    COUNT(*) AS cnt,
    MIN(CAST(ts_text AS TIMESTAMP_NTZ)) AS min_ts,
    MAX(CAST(ts_text AS TIMESTAMP_NTZ)) AS max_ts
  FROM logs.syslog_curated
  WHERE dt = '2026-09-15'
  UNION ALL
  SELECT
    'new_syslog_iceberg' AS src,
    COUNT(*) AS cnt,
    MIN(ts) AS min_ts,
    MAX(ts) AS max_ts
  FROM hive_prod.logs.syslog_iceberg
  WHERE dt = DATE '2026-09-15'
);

SELECT src, cnt, min_ts, max_ts
FROM (
  SELECT
    'old_authlog_curated' AS src,
    COUNT(*) AS cnt,
    MIN(CAST(ts_text AS TIMESTAMP_NTZ)) AS min_ts,
    MAX(CAST(ts_text AS TIMESTAMP_NTZ)) AS max_ts
  FROM authlogs.authlog_curated
  WHERE dt = '2026-09-15'
  UNION ALL
  SELECT
    'new_authlog_iceberg' AS src,
    COUNT(*) AS cnt,
    MIN(ts) AS min_ts,
    MAX(ts) AS max_ts
  FROM hive_prod.logs.authlog_iceberg
  WHERE dt = DATE '2026-09-15'
);
EOF

件数と時刻範囲に加え、host、program、severity、dt、hrの分布も確認する。

sudo -u spark /usr/local/bin/spark-sql-iceberg <<'EOF'
SET spark.sql.hive.convertMetastoreParquet=false;
SET spark.sql.session.timeZone=Asia/Tokyo;

WITH old_data AS (
  SELECT host, program, severity,
         CAST(dt AS DATE) AS dt,
         HOUR(CAST(ts_text AS TIMESTAMP_NTZ)) AS hr,
         COUNT(*) AS cnt
  FROM logs.syslog_curated
  WHERE dt = '2026-09-15'
  GROUP BY host, program, severity, CAST(dt AS DATE), HOUR(CAST(ts_text AS TIMESTAMP_NTZ))
), new_data AS (
  SELECT host, program, severity, dt, hr, COUNT(*) AS cnt
  FROM hive_prod.logs.syslog_iceberg
  WHERE dt = DATE '2026-09-15'
  GROUP BY host, program, severity, dt, hr
), old_only AS (
  SELECT * FROM old_data
  EXCEPT
  SELECT * FROM new_data
), new_only AS (
  SELECT * FROM new_data
  EXCEPT
  SELECT * FROM old_data
)
SELECT 'old_only' AS diff, * FROM old_only
UNION ALL
SELECT 'new_only' AS diff, * FROM new_only;

WITH old_data AS (
  SELECT host, program, severity,
         CAST(dt AS DATE) AS dt,
         HOUR(CAST(ts_text AS TIMESTAMP_NTZ)) AS hr,
         COUNT(*) AS cnt
  FROM authlogs.authlog_curated
  WHERE dt = '2026-09-15'
  GROUP BY host, program, severity, CAST(dt AS DATE), HOUR(CAST(ts_text AS TIMESTAMP_NTZ))
), new_data AS (
  SELECT host, program, severity, dt, hr, COUNT(*) AS cnt
  FROM hive_prod.logs.authlog_iceberg
  WHERE dt = DATE '2026-09-15'
  GROUP BY host, program, severity, dt, hr
), old_only AS (
  SELECT * FROM old_data
  EXCEPT
  SELECT * FROM new_data
), new_only AS (
  SELECT * FROM new_data
  EXCEPT
  SELECT * FROM old_data
)
SELECT 'old_only' AS diff, * FROM old_only
UNION ALL
SELECT 'new_only' AS diff, * FROM new_only;
EOF

結果が0行であれば、この粒度では差分がない。複数日で比較し、特に日付境界の00時台と23時台でdt、hr、tsが旧方式と一致することを確認する。

curatedテーブルの廃止

比較が完了するまでは旧バッチを止めず、curatedもDROPしない。移行は次の順で行う。

1. 旧curated生成を継続したまま直接ロードを手動実行する
2. 複数日の件数、MIN(ts)、MAX(ts)、各分布を比較する
3. 日次ロードをload_raw_to_iceberg.shへ切り替える
4. 翌日の定期実行と後続のcompaction、snapshot cleanupを確認する
5. 問題がなければcurated生成cronを停止する
6. 一定期間戻せる状態で監視した後、Hiveオブジェクトとcurated実体を廃止する

今回の直接ロード経路で不要になるものは次である。

不要
├─ logs.syslog_curated
├─ authlogs.authlog_curated
├─ logs.v_syslog_parsed_day
├─ authlogs.v_authlog_parsed_day
├─ logs.syslog_raw_day
├─ authlogs.authlog_raw_day
├─ /usr/local/share/syslog_init.sql
├─ /usr/local/share/syslog_load_dt.sql
├─ /usr/local/share/authlog_init.sql
├─ /usr/local/share/authlog_load_dt.sql
├─ /usr/local/sbin/syslog_curate_rotate.sh のcurated生成処理
├─ /usr/local/sbin/authlog_curate_rotate.sh のcurated生成処理
└─ /opt/iceberg/bin/load_curated_to_iceberg.sh

継続使用
├─ rsyslog / Fluentd
├─ Kafka
├─ Kafka Connect HDFS Sink
├─ HDFS raw
├─ Spark
├─ /usr/local/bin/spark-sql-iceberg
├─ Apache Iceberg
├─ Hive Metastore
├─ Trino
└─ Icebergのcompaction / snapshot / orphan fileメンテナンス

syslog_curate_rotate.shとauthlog_curate_rotate.shにはrawの7日ローテーションも含まれている。スクリプト全体を単純に削除するとrawローテーションも止まるため、raw削除処理は別ジョブとして残す必要がある。

また、既存のHDFS比較・再取り込み処理はsyslog_curatedの件数比較とsyslog_curate_rotate.shの再実行を前提としている。これも直接ロード後は、HDFS rawの同期後にload_raw_to_iceberg.sh YYYY-MM-DDを呼ぶよう変更対象になる。

変更後のデータフロー確認

確認する。

sudo -u spark /usr/local/bin/spark-sql-iceberg <<'EOF'
SET spark.sql.session.timeZone=Asia/Tokyo;

SELECT dt, COUNT(*) AS cnt, MIN(ts) AS min_ts, MAX(ts) AS max_ts
FROM hive_prod.logs.syslog_iceberg
GROUP BY dt
ORDER BY dt DESC
LIMIT 7;

SELECT dt, COUNT(*) AS cnt, MIN(ts) AS min_ts, MAX(ts) AS max_ts
FROM hive_prod.logs.authlog_iceberg
GROUP BY dt
ORDER BY dt DESC
LIMIT 7;
EOF

将来のS3化について

今回はS3化しない。現在の保存先はHDFSのままである。

今回Hive curatedへの依存を外すことで、将来rawとIcebergの保存先をS3互換Object StorageまたはAmazon S3へ変更するときも、Sparkの入力パスとIceberg catalog/warehouseを中心に検討できる。S3の具体的な構築は別記事の範囲とする。

まとめ

Kafka Connect HDFS Sinkが保存したJSON TextをSpark SQLのtextデータソースで直接読み、既存Hive ViewにあったJSON抽出、timestamp整形、severity判定、dtとhrの生成を移植した。

時刻は既存設計を維持し、rawのローカル時刻文字列をTIMESTAMP_NTZへ変換した。追加のUTC/JST補正は行っていない。

これにより、日次処理は次の流れになる。

curatedをすぐ削除せず、旧方式と新方式を複数日比較してから切り替えることが重要である。また、curated生成スクリプトに同居していたrawローテーションは、廃止せず別ジョブへ分離する必要がある。

関連記事

• 【導入編】HDFSログ分析基盤をAWS S3へ段階移行する全8編の構成を紹介する
• 【第2編】HDFSのrawログとApache IcebergをローカルS3へ移行してみる
• 各ホストのsyslogを収集して、bigtop基盤で可視化してみた。
• 列指向データ作成,ローテーションスクリプト作成してみた
• Hadoop収集対象にauth.log追加してみた
• ログ収集・解析基盤にSpark/Icebergを構築してみた
• Kafka / Spark / Flink / Iceberg で時刻(ts)の扱いを統一する設計

実機検証について

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

実施する場合は、検証環境でbackupとrollback手順を用意し、旧方式との件数、時刻範囲、集計結果を比較してから本番環境へ適用する。実機検証の結果に応じて本記事を更新する予定である。