ServiceNowのデータをDataLakeへ寄せてみよう！ Iceberg変換してみた手順まとめ

2026年4月23日 08:17

ここ最近の何本かのブログは、ServiceNowのデータ分析に関連する記事を投稿してきました。

今回は、現在、分析専用データプラットフォーム基盤だけの範囲に留まらず、AI Agentの利用する "RAGデータが湧き出る泉" としても期待されている ”データレイク” のデータ構成へ ServiceNowのレコードデータを変換する記事を書いてみます。

その理由を挙げると、、以下のようなところです。

・ServiceNow内のデータ分析は様々な制約に縛られて、面倒
・データレイクに搭載されるMLやLLMは正規化の効率性を強烈に追求
・高速分散処理できるSparkなどでETL変換処理なども高速化が可能

早い話、”餅は餅屋に任せとこう！”　といったところで、利用できるのであれば、データレイクを使って分析およびRAGデータ整備の効率化を図っちゃおう！　というのが目的です。

１．環境準備

以下の開発環境を準備します。Pythonのインストールはこちら、Databricksのアカウント取得はこちらを参照ください。
　　・AustraliaバージョンPDI
　　・Python3.1X以上
　　・Databricks無償トライアルアカウント

プロジェクトフォルダを用意します。
例）C:\work\sn_iceberg_db

このフォルダをcurrentフォルダとしてプログラムの配置を行います。

続いて、Pythonのライブラリのインストールを一括で行います。

※ライブラリのインストールが失敗する場合はPythonバージョンとの互換性の問題です。最新のPython環境の場合、インストールが失敗する場合があります。その場合には、後続バージョンを指定します。

pip install "pyiceberg[pyarrow,sql-sqlite]==0.8.1" "pyarrow>=17.0.0" "requests==2.31.0" "python-dotenv==1.0.0"

.envファイルを↑のプロジェクトフォルダに作成します。

この.envファイルにPDIの接続情報を記載して、Pythonコードステップで読み取れるように準備します。

SN_INSTANCE=https://<your-instance>.service-now.com
SN_USER=admin
SN_PASSWORD=実際のパスワード

２．DataLakeのデータ構成アーキテクチャー

データレイクは「あらゆるデータを生のまま蓄積し、目的に応じて加工・活用する」ための基盤です。

従来のデータウェアハウス（DWH）が「整形済みデータだけを扱う」のに対し、データレイクは構造化・非構造化を問わずデータを受け入れ、分析・ML・LLM など多様な用途に対応します。

現代のデータレイクは以下のような多層構造（メダリオンアーキテクチャ）で設計されるのが一般的です。

特に近年注目されているのが コンシューマー層の最右翼に位置する「AI Agent / RAG」 です。LLM が正確な回答を返すためには、鮮度の高い・正規化された業務データを RAG（検索拡張生成）として供給する必要があります。データレイクはまさに 「RAGデータが湧き出る泉」 として機能します。

３．Icebergとはなんぞや？

3-1. データレイクの「テーブルフォーマット」問題

データレイクのストレージ層では、実データは Parquet や ORC などのファイル形式で保存されます。しかしファイルを置くだけでは以下の問題が起きます。

これを解決するのが テーブルフォーマット という概念であり、その代表実装が Apache Iceberg です。

Apache Icebergテーブルは、以下のような "metadata層 + data層" の二つの層から構成されるデータの保持様式を採用しています。

metadata 層こそが Iceberg の本質です。 Parquet ファイルをただ置くだけでは「ファイルの山」ですが、metadata 層が加わることで初めて「テーブル」になります。

そして、Iceberg は Bronze〜Gold のどの層にも適用できます。処理エンジン（Spark / Databricks / Athena / Snowflake）はすべて Iceberg を標準でサポートしており、ベンダーロックインなしにデータを持ち運べる点が最大の強みです。

４．ServiceNowからのデータを変換連携させる仕組み

ここでようやく ServiceNow との接続点が見えてきます。

ServiceNow のデータを Iceberg に変換することで得られるもの：

分析の自由度 → GlideRecord の制約を超え、Spark で大規模集計が可能
AI との接続 → 正規化・ベクトル化を DataLake 側で処理し、LLM へ供給
データの永続性 → ServiceNow のレコード変更・削除後もスナップショットで過去データを保持
オープン標準 → Databricks・Snowflake・AWS Athena など任意の処理エンジンで参照可能

次章からは、実際にこの変換処理を Python で実装していきます。

５．Incidentレコードの一括抽出からIceberg変換プログラムの作成

今回はPDIインスタンスより、Incidentテーブルレコードをバルクで抽出します。以下のPythonスクリプトファイルを作成します。

・01_extract.py

"""
STEP 1: ServiceNow incident テーブルからデータを取得してJSONに保存
"""

import requests
import json
import os
from dotenv import load_dotenv

load_dotenv()

SN_INSTANCE = os.getenv("SN_INSTANCE")
SN_USER     = os.getenv("SN_USER")
SN_PASSWORD = os.getenv("SN_PASSWORD")

# 取得するフィールドを絞る（Icebergスキーマをシンプルにするため）
FIELDS = [
    "sys_id",
    "number",
    "short_description",
    "description",
    "state",
    "priority",
    "category",
    "assigned_to",
    "opened_at",
    "resolved_at",
    "sys_created_on",
    "sys_updated_on",
]

def fetch_incidents(limit=100):
    url = f"{SN_INSTANCE}/api/now/table/incident"
    params = {
        "sysparm_limit": limit,
        "sysparm_fields": ",".join(FIELDS),
        "sysparm_display_value": "true",   # コード値ではなく表示名で取得
        "sysparm_exclude_reference_link": "true",
    }
    headers = {"Accept": "application/json"}

    print(f"[1/2] ServiceNow ({SN_INSTANCE}) に接続中...")
    response = requests.get(
        url,
        auth=(SN_USER, SN_PASSWORD),
        headers=headers,
        params=params,
    )
    response.raise_for_status()

    records = response.json().get("result", [])
    print(f"[2/2] {len(records)} 件のインシデントを取得しました")
    return records


def save_json(records, path="incidents.json"):
    with open(path, "w", encoding="utf-8") as f:
        json.dump(records, f, ensure_ascii=False, indent=2)
    print(f"保存完了: {path}")


if __name__ == "__main__":
    records = fetch_incidents(limit=100)
    save_json(records, "incidents.json")
    print("\n次のステップ: python 02_convert_iceberg.py を実行してください")

１）PDIのincidentデータを最大100件取得します。
２）FIELDS = [ ]：12個のフィールドをレコードデータとして取得します。
３）incidents.jsonファイルにJSON形式でレコードデータが構造化データとして保存されます。

続いて、このincident.jsonファイルをIceberg様式に変換するスクリプトを作成します。

・02_convert_incident.py

"""
STEP 2: incidents.json を Apache Iceberg 形式に変換してローカルに書き出す
【方式変更】PyIceberg SqlCatalog の Windows パスバグを回避するため、
  Parquet ファイルを PyArrow で直接書き出し、
  Iceberg メタデータ JSON を手動生成する方式に変更。
  依存ライブラリ: pyarrow のみ（pyiceberg 不要）
出力先: ./warehouse/  ← このフォルダをDatabricks Workspaceにアップロードする
"""

import json
import uuid
import time
import os
import pyarrow as pa
import pyarrow.parquet as pq

# ──────────────────────────────────────────────
# 定数
# ──────────────────────────────────────────────
DATABRICKS_BASE = (
    "/Workspace/Users/munenari.uraguchi@servicenow.com"
    "/servicenow_iceberg/warehouse"
)
DB_TABLE_DIR = f"{DATABRICKS_BASE}/servicenow.db/incident"

ARROW_SCHEMA = pa.schema([
    pa.field("sys_id",            pa.string(), nullable=False),
    pa.field("number",            pa.string(), nullable=True),
    pa.field("short_description", pa.string(), nullable=True),
    pa.field("description",       pa.string(), nullable=True),
    pa.field("state",             pa.string(), nullable=True),
    pa.field("priority",          pa.string(), nullable=True),
    pa.field("category",          pa.string(), nullable=True),
    pa.field("assigned_to",       pa.string(), nullable=True),
    pa.field("opened_at",         pa.string(), nullable=True),
    pa.field("resolved_at",       pa.string(), nullable=True),
    pa.field("sys_created_on",    pa.string(), nullable=True),
    pa.field("sys_updated_on",    pa.string(), nullable=True),
])


# ──────────────────────────────────────────────
# Step 1: データ読み込み
# ──────────────────────────────────────────────
def load_incidents(path="incidents.json"):
    with open(path, "r", encoding="utf-8") as f:
        records = json.load(f)
    print(f"[1/4] {len(records)} 件のインシデントを読み込みました")
    return records


# ──────────────────────────────────────────────
# Step 2: PyArrow Table に変換
# ──────────────────────────────────────────────
def records_to_arrow(records):
    columns = {field: [] for field in ARROW_SCHEMA.names}
    for rec in records:
        for field in ARROW_SCHEMA.names:
            val = rec.get(field)
            if isinstance(val, dict):
                val = val.get("display_value") or val.get("value") or ""
            columns[field].append(str(val) if val is not None else None)
    table = pa.Table.from_pydict(columns, schema=ARROW_SCHEMA)
    print(f"[2/4] PyArrow Table 変換完了: {table.num_rows} 行 × {table.num_columns} 列")
    return table


# ──────────────────────────────────────────────
# Step 3: Parquet + Iceberg メタデータを手動生成
# ──────────────────────────────────────────────
def write_iceberg_manual(arrow_table):
    local_base   = os.path.abspath("./warehouse")
    data_dir     = os.path.join(local_base, "servicenow.db", "incident", "data")
    metadata_dir = os.path.join(local_base, "servicenow.db", "incident", "metadata")
    os.makedirs(data_dir,     exist_ok=True)
    os.makedirs(metadata_dir, exist_ok=True)

    # ── 3a. Parquet ファイル書き出し ──
    parquet_filename = f"{uuid.uuid4()}.parquet"
    local_parquet    = os.path.join(data_dir, parquet_filename)
    pq.write_table(
        arrow_table,
        local_parquet,
        compression="snappy",
        write_statistics=True,
    )
    file_size = os.path.getsize(local_parquet)
    print(f"[3/4] Parquet 書き出し完了: {parquet_filename} ({file_size:,} bytes)")

    # ── 3b. Iceberg スキーマ (spec v2 JSON) ──
    iceberg_schema = {
        "type": "struct",
        "schema-id": 0,
        "fields": [
            {"id": 1,  "name": "sys_id",           "required": True,  "type": "string"},
            {"id": 2,  "name": "number",            "required": False, "type": "string"},
            {"id": 3,  "name": "short_description", "required": False, "type": "string"},
            {"id": 4,  "name": "description",       "required": False, "type": "string"},
            {"id": 5,  "name": "state",             "required": False, "type": "string"},
            {"id": 6,  "name": "priority",          "required": False, "type": "string"},
            {"id": 7,  "name": "category",          "required": False, "type": "string"},
            {"id": 8,  "name": "assigned_to",       "required": False, "type": "string"},
            {"id": 9,  "name": "opened_at",         "required": False, "type": "string"},
            {"id": 10, "name": "resolved_at",       "required": False, "type": "string"},
            {"id": 11, "name": "sys_created_on",    "required": False, "type": "string"},
            {"id": 12, "name": "sys_updated_on",    "required": False, "type": "string"},
        ]
    }

    # ── 3c. metadata.json 生成 ──
    now_ms      = int(time.time() * 1000)
    snapshot_id = int(time.time()) % (2 ** 31)
    meta_uuid   = str(uuid.uuid4())

    db_parquet_path  = f"file://{DB_TABLE_DIR}/data/{parquet_filename}"
    db_manifest_path = f"file://{DB_TABLE_DIR}/metadata/{uuid.uuid4()}-m0.avro"

    metadata = {
        "format-version": 2,
        "table-uuid": str(uuid.uuid4()),
        "location": f"file://{DB_TABLE_DIR}",
        "last-sequence-number": 1,
        "last-updated-ms": now_ms,
        "last-column-id": 12,
        "current-schema-id": 0,
        "schemas": [iceberg_schema],
        "default-spec-id": 0,
        "partition-specs": [{"spec-id": 0, "fields": []}],
        "last-partition-id": 999,
        "default-sort-order-id": 0,
        "sort-orders": [{"order-id": 0, "fields": []}],
        "properties": {
            "write.parquet.compression-codec": "snappy"
        },
        "current-snapshot-id": snapshot_id,
        "refs": {
            "main": {"snapshot-id": snapshot_id, "type": "branch"}
        },
        "snapshots": [
            {
                "sequence-number": 1,
                "snapshot-id": snapshot_id,
                "timestamp-ms": now_ms,
                "summary": {
                    "operation": "append",
                    "added-data-files": "1",
                    "added-records": str(arrow_table.num_rows),
                    "added-files-size": str(file_size),
                    "changed-partition-count": "1",
                    "total-records": str(arrow_table.num_rows),
                    "total-files-size": str(file_size),
                    "total-data-files": "1",
                    "total-delete-files": "0",
                    "total-position-deletes": "0",
                    "total-equality-deletes": "0"
                },
                "manifest-list": db_manifest_path,
                "schema-id": 0
            }
        ],
        "statistics": [],
        "snapshot-log": [
            {"timestamp-ms": now_ms, "snapshot-id": snapshot_id}
        ],
        "metadata-log": []
    }

    # ── 3d. metadata.json 保存 ──
    meta_filename = f"00000-{meta_uuid}.metadata.json"
    local_meta    = os.path.join(metadata_dir, meta_filename)
    with open(local_meta, "w", encoding="utf-8") as f:
        json.dump(metadata, f, indent=2, ensure_ascii=False)

    # ── 3e. version-hint.text（バージョン 0 を示す） ──
    with open(os.path.join(metadata_dir, "version-hint.text"), "w") as f:
        f.write("0")

    print(f"[4/4] Iceberg メタデータ生成完了: {meta_filename}")
    print(f"\n{'='*60}")
    print(f"ローカル出力先 : {local_base}")
    print(f"Databricks パス: {DB_TABLE_DIR}")
    print(f"レコード数     : {arrow_table.num_rows}")
    print(f"Parquet ファイル: {parquet_filename}")
    print(f"メタデータ     : {meta_filename}")
    print(f"{'='*60}")
    print("""
次のステップ（PowerShell）:

  databricks workspace import-dir `
    C:\\work\\sn_iceberg_db\\warehouse `
    /Users/munenari.uraguchi@servicenow.com/servicenow_iceberg/warehouse `
    --overwrite

アップロード確認:
  databricks workspace ls /Users/munenari.uraguchi@servicenow.com/servicenow_iceberg/warehouse/servicenow.db/incident/data
  databricks workspace ls /Users/munenari.uraguchi@servicenow.com/servicenow_iceberg/warehouse/servicenow.db/incident/metadata
""")

    return local_base


# ──────────────────────────────────────────────
# Main
# ──────────────────────────────────────────────
if __name__ == "__main__":
    records   = load_incidents("incidents.json")
    arrow_tbl = records_to_arrow(records)
    write_iceberg_manual(arrow_tbl)

１）標準pyicebergライブラリを使わずpyarrowでiceberg様式へ変換
→後述しますが、当方のWindows環境で起きた不具合で、成果物内のファイルパス文字列がうまく生成できませんでした。切り替えてます。
２）load_incidents(path="incidents.json"):メモリ上にincident.jsonをロード
３）records_to_arrow(records):JSONの各レコードをPyArrow形式の表構造に変換
４）catalog.create_namespace()：カタログの生成。今回はローカルで完結するSQLiteをカタログのバックエンドに使用。
５）table = catalog.create_table()：Iceberg形式のファイル群を生成。
以下のようなファイル群が構成。

warehouse/
└── servicenow/
    └── incident/
        ├── data/         ← Parquetファイル（実データ）
        └── metadata/     ← Icebergのメタデータ（JSON/Avro）

この中のmetadata/ こそがIcebergの本質で、スキーマ履歴・スナップショット・パーティション情報などが管理されます。

これで、ほぼ本要件を動作させる準備は整いました。

６．レコードの取得からIcebergデータへ一括変換を実施

では、さっそく動かして確認してみましょう。

①PDIからIncidentレコードの一括取得
作業フォルダ上にて、以下ステップを実行します。

python 01_extract.py

実行すると、以下のような出力と共に

・incidents.json

が作業フォルダ上に作成されます。

中身を開くと、以下のようなIncidentレコードがJSONで格納されています。
・incidents.jsonファイル内のJSONデータ

[
  {
    "sys_id": "1c741bd70b2322007518478d83673af3",
    "number": "INC0000060",
    "short_description": "Unable to connect to email",
    "opened_at": "2016-12-12 07:19:57",
    "resolved_at": "2016-12-13 13:43:14",
    "sys_created_on": "2016-12-12 07:19:57",
    "description": "I am unable to connect to the email server. It appears to be down.",
    "state": "Closed",
    "sys_updated_on": "2016-12-13 18:46:44",
    "priority": "3 - Moderate",
    "category": "Inquiry / Help",
    "assigned_to": "David Loo"
  },
  {
    "sys_id": "1c832706732023002728660c4cf6a7b9",
    "number": "INC0009002",
    "short_description": "My computer is not detecting the headphone device",
    "opened_at": "2018-09-16 05:49:23",
    "resolved_at": "2018-09-16 05:51:17",
    "sys_created_on": "2018-09-16 05:50:05",
    "description": "My computer is not detecting the headphone device. It could be an issue with the USB port.",
    "state": "Closed",
    "sys_updated_on": "2025-03-06 02:55:55",
    "priority": "3 - Moderate",
    "category": "Hardware",
    "assigned_to": ""
  },
　　（以下省略）

Incidentのレコード取得はできています。つづいて、以下スクリプトを実行し、Iceberg様式へレコードデータを一括変換します。

python 02_convert_iceberg.py

実行すると、以下のような出力と共に

以下ファイル群が作成されています。

warehouse/
└── servicenow/
    └── incident/
        ├── data/
        │   └── *.parquet    ← 実データ（バイナリ）
        └── metadata/
            ├── *.avro       ← バイナリファイル（スナップショット情報）
            ├── *.json       ← スキーマ・パーティション情報（テキスト）
            └── version-hint.text ← 現在のバージョン番号

Avroファイルを開くと、中身はバイナリデータですね。

Icebergのメタデータは .avro 形式（バイナリ）で保存されます。テキストファイルではないのでVSCodeのようなエディタで開くと文字化けして見えますが、中身は正しく書かれています。

実際のデータはparquetというファイルに格納されています。これもバイナリなのでエディタで直接開けないので、以下pythoonコマンド実行で開きます。

PyArrow + pandasでparquetファイルを参照します。

PS C:\work\sn_iceberg_db> python -c "import pyarrow.parquet as pq; t = pq.read_table('warehouse/servicenow.db/incident/data/00000-0-5af1299b-7d5c-4004-ae7f-c0ce49e8bac9.parquet'); print(t.to_pandas())"
                              sys_id      number  ...       sys_created_on       sys_updated_on
0   1c741bd70b2322007518478d83673af3  INC0000060  ...  2016-12-12 07:19:57  2016-12-13 18:46:44
1   1c832706732023002728660c4cf6a7b9  INC0009002  ...  2018-09-16 05:50:05  2025-03-06 02:55:55
2   46b66a40a9fe198101f243dfbc79033d  INC0000009  ...  2024-12-16 14:51:33  2025-03-05 12:16:04
3   46b9490da9fe1981003c938dab89bda3  INC0000010  ...  2024-12-16 14:54:41  2025-03-05 12:16:35
4   46c03489a9fe19810148cd5b8cbf501e  INC0000011  ...  2024-12-16 15:02:15  2025-03-05 11:56:31
..                               ...         ...  ...                  ...                  ...
62  ed92e8d173d023002728660c4cf6a7bc  INC0009005  ...  2018-08-31 21:35:45  2018-12-12 23:18:55
63  ef4225a40a0a0b5700d0b8a790747812  INC0000049  ...  2025-02-12 10:05:40  2025-03-05 11:52:34
64  ef43c6d40a0a0b5700c77f9bf387afe3  INC0000050  ...  2025-02-12 10:05:40  2025-03-05 11:49:40
65  f12ca184735123002728660c4cf6a7ef  INC0007001  ...  2018-10-16 22:47:45  2025-03-06 02:45:53
66  ff4c21c4735123002728660c4cf6a758  INC0007002  ...  2018-10-16 22:48:24  2018-12-12 23:28:49

これがIncidentレコードの本体ですね。

このように、標準pyiceberg機能を使い、
・metadata と dataのIceberg 様式
に変換が成功しました！

これで、DataLake側より参照やデータ操作が行えるようになりました。

７．Databricksへフォルダ構成をアップロードして確認

本来はDatabricksのUnity Catalogを介すとこうしたデータアクセスが簡単に行えるとのことですが、、残念ながら個人利用の無償環境ではUnity Catalogが利用できないため、今回はダイレクトアップロードで確認を行います。

個人利用のFreeEditionではUIからのアップロード機能がないようです。

DatabricksのCLIライブラリをインストールします。

winget install Databricks.DatabricksCLI

※Pythonのインストーラも使ってみたのですが、うまく入りませんでした。wingetを使って最新CLIを入れてみました。

うまくインストールできると、以下コマンドでバージョンが確認できます。

PS C:\work\sn_iceberg_db> databricks version
Databricks CLI v0.297.2
PS C:\work\sn_iceberg_db>

続いて、アクセス用のTokenKeyを払い出します。
・[設定] → [開発者] のアクセストークンメニュー

ここで新規でアクセストークンの作成を行います。

・コメント：任意文字列
・存続期間：90日
・APIスコープ：すべてのAPI　※検証用途なのでこ知らを選択

では、コマンドで接続します。

databricks configure --host https://[ホスト名].cloud.databricks.com --token

※ホスト名には、自分用に配布されたホスト名をブラウザのURLアドレスからコピーして入力します。

続いて、Token入力が求められます。先の手順で取得したTokenキー文字列を入力し、Enterキーを押下します。

v Personal access token:

応答はないですが、続けて、以下コマンドで接続を確認しましょう。
・databricks fs ls dbfs:/

PS C:\work\sn_iceberg_db> databricks fs ls dbfs:/                                                           
Volumes
Workspace
databricks-datasets
PS C:\work\sn_iceberg_db>

既存の構成が参照できました。

では、ここからIceberg変換されたフォルダ一式をアップロードします。

databricks workspace import-dir `
  C:\work\sn_iceberg_db\warehouse `
  /Users/your-email@example.com/servicenow_iceberg/warehouse `
  --overwrite

１）workspaceにアップロード
２）アカウント登録時のメールアドレスを指定
３）DBFSではなくWorkspaceを使う理由：Free EditionはPublic DBFS rootが無効化されていたため。

アップロードが成功すると以下が応答されます。

Importing files from C:\work\sn_iceberg_db\warehouse
servicenow.db\incident\data\xxxxxxxx.parquet -> /Users/.../data/xxxxxxxx.parquet
servicenow.db\incident\metadata\00000-xxxxxxxx.metadata.json -> /Users/.../metadata/00000-xxxxxxxx.metadata.json
servicenow.db\incident\metadata\version-hint.text -> /Users/.../metadata/version-hint.text
Import complete

Databricksのメニューの[新規]→[ノートブック]より、ノートブックを作成します。

以下コマンドを最初のセルに貼り付けます。

%pip install "pyiceberg[pyarrow,sql-sqlite]" pyarrow --quiet

実行終了すると、

Kernel再起動の警告がでてます。コマンド用のセルを追加しましょう。

以下コマンドを貼り付けて、再起動させます。

dbutils.library.restartPython()

完了。

さあ、いよいよIceberg変換済データを読み込みます！
※実はここに辿り着くまで、無償版での制約に引っかかりながら何度もトライアンドエラーを繰り返してました💦

Databricks の Spark は /Workspace/... パスを直接読めないため、PyArrow + pandas を使います。

新しいセルを作って、以下コマンドを実行します。

import pyarrow.parquet as pq

parquet_path = (
    "/Workspace/Users/your-email@example.com"
    "/servicenow_iceberg/warehouse"
    "/servicenow.db/incident/data"
    "/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx.parquet"  # 実際のファイル名に変更
)

table = pq.read_table(parquet_path)
df = table.to_pandas()
print(f"レコード数: {len(df)}")
display(df)

１）個人登録用メールアドレス
２）実際のファイル名を入力

結果が応答されました。完璧です！67件のレコードが表示されました。

続けて、Priority 別の件数を集計する次のコマンドを実行します。

result = (
    df.groupby("priority")["sys_id"]
    .count()
    .reset_index()
)
result.columns = ["priority", "count"]
result = result.sort_values("count", ascending=False)
display(result)

優先度ごとの集計ができてますね！

カテゴリ別でも集計してみましょう。

result = (
    df.groupby("category")["sys_id"]
    .count()
    .reset_index()
    .sort_values("sys_id", ascending=False)
)
result.columns = ["category", "count"]
display(result)

はい。確かにカテゴリ別でも集計されています！

今回の試みにて、ServiceNowのIncidetnレコードを対象に、Iceberg変換できました！

これでDataLake側にデータを寄せて分析やデータ正規化を処理させるアプローチが取れることが実証できました！

８．ハマりポイントまとめ

1. PyIceberg on Windows のパスバグ

今回の中継・加工処理（ETL)をWindows端末で実施したのが不具合が混入された要因。

SqlCatalog の warehouse に file:///C:/work/... を渡すと、PyArrow が /C:/work/... と解釈してエラーになります。今回は PyIceberg を使わず PyArrow で直接 Parquet + メタデータを生成することで回避しました。

※AWSなどでは、別のI/Oパスを使うためこのバグは関係しない。DataricksもJVM上でIcebergをネイティブ変換するのでこの不具合は関係ない

2. Spark から /Workspace/ パスを直接読めない

Databricks の Spark は /Workspace/ パスを dbfs:/Workspace/... として解釈しますが、Workspace ファイルへの Spark 直接アクセスは制限されています。pq.read_table() で PyArrow 経由で読むことで解決しました。

3. Databricks CLI の import-dir はバイナリファイルも転送できる

.parquet や .db などのバイナリファイルも --overwrite オプションで上書きアップロードできます。

９．まとめ

如何でしたでしょうか？

ServiceNow PDI のインシデントデータを Apache Iceberg（Parquet + メタデータ JSON）に変換し、Databricks で分析するまでの手順を紹介しました。

今回は、、本実装時には起こりえない、「Windows端末上でデータ変換処理（ETL）」をさせていたので、PyIceberg の Windows 環境バグにぶちあたりました💦

これの回避策として、「PyArrow で直接生成する」方式を採用しましたが、”PyIcebergライブラリによる変換”が王道パータンな点、コメント補足しておきます。

次回もAI Agentに繋がる技術をご案内したいと思います。では。