はじめに
Databricks Japanのハンズオン教材を集めたGitHubリポジトリに、非構造化データパイプラインのハンズオンを追加しました。
音声・画像・帳票という形式の違うファイルを、Databricks上で1本のパイプラインにまとめ、分析や業務に使えるテーブルにするという内容です。ノートブックを00から順に実行すれば一通り動くように作っているので、この記事はその歩き方のガイドを兼ねています。
ハンズオンで持ち帰ってほしいことは2つです。
- 流れは共通。形式が違っても「取り込む → AIで取り出す → 確かめる → つなぐ」という同じ流れで処理できる。形式ごとに違うのは「AIで取り出す」に使う手段だけ
- そのまま運用に乗る。新しく届いたファイルだけを処理する作り(増分処理)にしておけば、処理済みのファイルにもう一度課金せず、同じノートブックをジョブにして自動で回せる
形式ごとに手段を選び分けた理由や、作っていて気づいた点も含めて書いていきます。
題材: 損害保険の事故受付
題材は損害保険の事故受付です。事故1件(受付番号 case_id)ごとに、次の3つのファイルが届くという想定にしました。
| 形式 | 中身 |
|---|---|
| 音声 | 事故受付の通話録音(WAV) |
| 画像 | 損害箇所の写真(JPEG) |
| 帳票 | 修理工場の見積書(PDFと、手書きの見積書をスキャンしたJPEG) |
3つから取り出した内容を事故1件＝1行のテーブルにまとめ、1つの形式だけを見ていては分からない食い違いを見つけます。「通話では左前をこすったと言っているのに、写真と見積書では右後ろが壊れている」「ほんの少しこすっただけという申告なのに、見積が高額」といったものです。
たとえばcase_03は、通話では「左前のバンパーを擦った」と申告している案件です。一緒に届く写真と見積書はこうなっています。
写真では、右後ろのバンパーに擦り傷があります。
見積書の損傷部位も「右リヤバンパー」です。通話・写真・見積書を別々に見ているとそれぞれは筋が通っているのに、3つを並べると申告だけが食い違っている。こういう案件を見つけるのがこのハンズオンのゴールです。通話音声は記事には載せられないので、リポジトリの data/audio/ で聞いてみてください。
サンプルデータは7案件分を同梱しています。会社名・人名・修理工場・番号はすべて架空で、音声はOpen JTalkによる合成音声、写真は画像生成モデルで作ったものです。
全体の構成
全体像はこうなります。
Volumeに届いた3形式のファイルを1つのBronzeテーブルに取り込み、形式ごとのSilverで中身を取り出し、Goldで事故1件＝1行にまとめます。Goldの案件の一覧はBronzeから作るので、Bronzeから直接つながる線があります。すべて実行したあとにUnity Catalogで bronze_files のリネージを開くと、同じつながりが表示されます。
ノートブックは次のとおりです。メダリオンアーキテクチャ(Bronze → Silver → Gold)に沿っています。
| # | ノートブック | 流れのどこか | 内容 |
|---|---|---|---|
| 0 | 00_setup |
— | 自分のスキーマとVolumeを作り、1回目のデータ(case_01〜05)を置く |
| 1 | 01_bronze_ingest |
取り込む | 3形式のファイルをAuto Loaderで1つのテーブルに取り込む |
| 2 | 02_silver_audio |
取り出す | 通話音声をGeminiに直接入力し、申告内容をJSONで受け取る |
| 3 | 03_silver_image |
取り出す | 損害写真から損傷の位置・部位・程度を ai_query で取り出す |
| 4 | 04_silver_form |
取り出す・確かめる | 見積書を ai_parse_document → ai_extract で読み、検算で人が確認すべきものを振り分ける |
| 5 | 05_gold_join |
つなぐ | 3つを case_id でつなぎ、食い違いを見つける |
| 6 | 06_add_new_files |
運用に乗る | 2回目のデータ(case_06・07)を置き、01〜05を再実行して新しい2件だけが処理されることを確かめる |
| 付録 | appendix_evaluation |
確かめる | 見積書から取り出した値を正解データと突き合わせ、項目ごとの正解率を出す |
形式ごとの処理手段と課金をまとめるとこうなります。
| 形式 | 手段 | 課金 |
|---|---|---|
| 音声 | 基盤モデルAPIをPythonから呼ぶ(databricks-gemini-3-1-pro) |
トークン課金 |
| 画像 |
ai_query(databricks-gemini-3-5-flash) |
バッチ推論の対応モデルなので、バッチ推論として時間課金 |
| 帳票 |
ai_parse_document → ai_extract
|
AI Functionsの課金(ページ数・入力件数に応じたDBU) |
3形式とも「Bronzeから読んでSilverに書く」という形は同じで、真ん中の手段だけが違う。この表がハンズオン全体の要約になっています。
事前準備
前提条件
- Unity Catalogが有効なワークスペース(動作確認はAWS東京リージョンで行っています)
- サーバーレスのノートブックが使えること
- パートナーが提供するAI機能が有効であること(Geminiを使うため)
-
databricks-gemini-3-1-proとdatabricks-gemini-3-5-flashを呼び出せること - AI Functions(
ai_query・ai_parse_document・ai_extract)が使えるリージョンであること(リージョン別の機能の対応) - スキーマを作れるカタログ(
USE CATALOGとCREATE SCHEMAの権限)。既定はworkspace
リポジトリをGitフォルダとして取り込む
リポジトリをワークスペースにGitフォルダとして取り込みます。ノートブックは同じフォルダの data/ からサンプルデータを読むので、ノートブックだけをインポートするのではなく、リポジトリごと取り込んでください。
使うカタログが workspace 以外のときは、notebooks/_config の DEFAULT_CATALOG だけを書き換えます。スキーマはログインしているユーザー名から自動で作るので(yamada.taro@example.com → ws_yamada_taro)、複数人で同時に実行しても互いのテーブルは混ざりません。
# ===== 変更してよいのはここだけ =====
DEFAULT_CATALOG = "workspace" # スキーマを作る権限(USE CATALOG・CREATE SCHEMA)があるカタログ
# ====================================
ai_parse_documentが使える環境かを先に確かめる
ここが最初のハマりどころです。04で使う ai_parse_document は、公式ドキュメントではサーバーレス環境のバージョン3以上が要件とされています。古い環境のまま進むと、01〜03は動くのに04で初めてエラーになります。
そこで _config では、読み込まれるたびにNULLを渡して関数が使えるかだけを確かめ、使えなければその場で止めるようにしました。
try:
# 結果は真偽値にして受け取る(古い環境のノートブックはVARIANT型の値をそのまま受け取れないため)
spark.sql("SELECT ai_parse_document(CAST(NULL AS BINARY)) IS NULL AS ok").collect()
except Exception as e:
raise RuntimeError(
"このノートブックの環境では ai_parse_document が使えません。"
"右側の「環境」パネルの「基本環境」で新しいバージョンを選び、「適用」してから実行し直してください。"
) from e
止まったときは、右側の 環境 パネルの 基本環境 で新しいバージョンを選んで 適用 します(サーバーレス環境の設定)。ハンズオンの途中で全員が04で止まるより、最初の00で止まるほうがずっと立て直しやすい、という判断です。
取り込む: 3形式を1つのBronzeテーブルに
01_bronze_ingest では、Auto Loaderの binaryFile 形式で、landing/ の下にある音声・画像・帳票をまとめて1つのテーブルに取り込みます。この段階では中身を解釈せず、LLMも呼びません。
query = (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "binaryFile") # ファイルを中身(バイナリ)ごと1行として読む
.load(LANDING)
.select(
"path",
F.regexp_extract("path", r"/landing/(audio|image|form)/", 1).alias("modality"), # フォルダ名＝形式
F.regexp_extract("path", r"(case_\d+)", 1).alias("case_id"), # ファイル名から受付番号
"length",
"content",
F.current_timestamp().alias("ingested_at"),
)
.writeStream
.option("checkpointLocation", f"{CHECKPOINT}/bronze") # 処理済みのファイルの記録。消すと全件取り込み直しになる
.trigger(availableNow=True) # 今あるファイルを処理して止まる
.toTable(BRONZE)
)
query.awaitTermination()
trigger(availableNow=True) なので、常駐するストリーミングではなく、ふつうのバッチ処理と同じように終わります。
形式を問わず最初に1つのテーブルへ入れておくと、この後の工程はすべて「Bronzeから読む」という同じ入口になります。加えて、何が何件どのくらいの量届いたかをSQLで数えられる。本番で大量のファイルを扱うときは、この棚卸しが処理時間と費用を見積もる出発点になります。
AIで取り出す: 形式ごとに手段を選ぶ
ここからが本題です。3形式とも「Bronzeの未処理分を選ぶ → AIで取り出す → Silverに追記する」という形は同じですが、取り出す手段は形式ごとに変えています。
音声: 基盤モデルAPIに直接入力する
通話音声は、文字起こしの工程を挟まずにGemini(databricks-gemini-3-1-pro)へ直接入力し、申告内容をJSONで受け取ります。
音声だけPythonから呼んでいるのは、SQLの ai_query に files で渡せるファイルが画像(JPEG/PNG)だからです。音声を入力できるモデルは、基盤モデルAPIをOpenAI互換のクライアントから呼べば使えます(音声・動画モデルの照会)。
接続先とトークンはノートブックの実行情報から取るので、個人アクセストークンを発行する必要はありません。
%pip install -q openai
dbutils.library.restartPython()
import base64
from openai import OpenAI
ctx = dbutils.notebook.entry_point.getDbutils().notebook().getContext()
client = OpenAI(api_key=ctx.apiToken().get(), base_url=f"{ctx.apiUrl().get()}/serving-endpoints")
def analyze_audio(content: bytes) -> str:
resp = client.chat.completions.create(
model=AUDIO_MODEL,
messages=[{"role": "user", "content": [
{"type": "text", "text": PROMPT},
# 音声はbase64の文字列にして audio_url として渡す
{"type": "audio_url", "audio_url": {"url": "data:audio/wav;base64," + base64.b64encode(content).decode()}},
]}],
# strict: スキーマどおりのJSONだけを返させる
response_format={"type": "json_schema", "json_schema": {"name": "accident_call", "schema": AUDIO_SCHEMA, "strict": True}},
)
c = resp.choices[0].message.content
# モデルによっては本文が「パーツの配列」で返るため、文字列の部分だけをつなぐ
return c if isinstance(c, str) else "".join(p.get("text", "") for p in c if isinstance(p, dict))
最後の行は一見冗長ですが、モデルを差し替えたときに message.content が文字列ではなく配列で返ってきても壊れないようにするためのものです。
AUDIO_SCHEMA では、損傷の位置・部位・程度に enum を付けています。この選択肢は写真・見積書と同じものを _config で定義して共有しています。05で3つを突き合わせるので、ここで語彙をそろえておかないと「左前」と「左フロント」のような表記の揺れを後で吸収する羽目になります。左右の基準(車の進行方向)も指示文に書いておきます。
DAMAGE_POSITIONS = ["左前", "右前", "左後ろ", "右後ろ", "前", "後ろ"]
DAMAGE_PARTS = ["フロントバンパー", "リヤバンパー", "フロントドア", "リヤドア", "フロントガラス", "その他"]
SEVERITIES = ["軽微", "中程度", "大きい"]
1件で試すと、こういうJSONが返ってきます。
{
"accident_datetime": "今朝、10月6日の朝8時半頃",
"location": "家の近くの交差点",
"situation": "信号待ちをしていたら、後ろから追突された。",
"injury": "首に少し違和感があり、念のため病院に行く予定。",
"counterparty": "あり",
"damage_position": "後ろ",
"damage_part": "リヤバンパー",
"damage_severity": "中程度",
"repair_shop": "緑自動車",
"summary": "契約者が信号待ち中に追突され、首に違和感があるため病院へ行く予定。車の後ろのバンパーがへこんでおり、緑自動車に修理の見積もりを依頼する。"
}
オペレーターと契約者の会話から、項目ごとに分かれて取り出せています。
本番の処理では4件ずつ並行して呼び、1件失敗しても全体は止めずにエラー内容を error 列に残します。
画像: ai_queryならSQLだけで書ける
写真は ai_query に files => content で直接渡せるので、SQLだけで書けます。responseFormat でJSONスキーマを指定し、failOnError => false で1件の失敗が全体を止めないようにしています。
SELECT b.case_id, b.path,
ai_query('databricks-gemini-3-5-flash', '<指示文>',
files => b.content, -- Bronzeのcontent列をそのまま渡す
responseFormat => '<JSONスキーマ>',
failOnError => false) AS r -- 結果は result と errorMessage の組で返る
FROM bronze_files b LEFT ANTI JOIN silver_image s ON b.path = s.path -- 未処理の写真だけ
WHERE b.modality = 'image'
課金は呼び出すモデルで変わります。databricks-gemini-3-5-flash はai_queryによるバッチ推論の対応モデルなので、バッチ推論として時間課金になります。対応モデル以外ならトークン課金です。
また、responseFormat に対応しないモデルもあります。モデルを替えるときは構造化出力の対応状況もあわせて確かめてください。
写真1枚が1行になり、損傷の位置・部位が選択肢の中から返ってきます。冒頭のcase_03の写真は「右後ろ」「リヤバンパー」と読めています。
帳票: ai_parse_document → ai_extract
見積書は、ai_parse_documentで文書を読み、その結果をai_extractに渡して項目を取り出します。帳票の読み取りと項目の取り出しに特化した関数なので、指示文を書かずにスキーマだけで済みます。PDFも、手書きをスキャンしたJPEGも同じ関数で読めます。
スキーマの description は取り出し方の指示として使われるので、正規化の指定もここに書きます。
EXTRACT_SCHEMA = json.dumps({
"estimate_no": {"type": "string", "description": "見積番号"},
"estimate_date": {"type": "string", "description": "見積日。西暦の YYYY-MM-DD 形式に直す(和暦の R8 は 2026 年)"},
"customer": {"type": "string", "description": "宛名の氏名(「様」は含めない)"},
"damage_position": {"type": "enum", "labels": DAMAGE_POSITIONS, "description": "損傷の位置。左右は車の進行方向が基準"},
"items": {"type": "array", "description": "明細の各行", "items": {"type": "object", "properties": {
"name": {"type": "string", "description": "作業内容・部品名"},
"amount": {"type": "integer", "description": "金額(円、税抜)"}}}},
"subtotal": {"type": "integer", "description": "小計(円)"},
"tax": {"type": "integer", "description": "消費税(円)"},
"total": {"type": "integer", "description": "合計金額(税込、円)"},
# 実際のノートブックでは事故日・工場名・損傷部位も取り出している
}, ensure_ascii=False)
ai_extract のスキーマでは、列挙は {"type": "enum", "labels": [...]} と書きます。ai_query の responseFormat(JSON Schemaの "enum": [...])とは書き方が違うので、写真の処理のスキーマを流用するときは書き換えが必要です。
呼び出しは1つのSQLで、version 2.1を指定して項目ごとの信頼度も返させています。
SELECT b.case_id, b.path,
ai_extract(ai_parse_document(b.content), '<スキーマ>',
map('version', '2.1', 'enableConfidenceScores', 'true')) AS ex
FROM bronze_files b LEFT ANTI JOIN silver_form s ON b.path = s.path
WHERE b.modality = 'form'
結果はVARIANT型で返るので、ex:response:shop:value のように : でたどって値を取り出します。信頼度は ex:response:shop:confidence_score です。明細の金額は aggregate で足し上げます。
aggregate(cast(ex:response:items AS ARRAY<VARIANT>), 0L,
(acc, i) -> acc + coalesce(i:amount:value::BIGINT, 0)) AS items_sum
和暦の「R8.10.27」が 2026-10-27 に直って返ってくるのは、スキーマの説明に書いた指定が反映された結果です。
確かめる: 信頼度ではなく業務のルールで振り分ける
取り出した値をそのまま使ってよいかを、どう判断するか。ai_extract の信頼度を閾値にしたくなりますが、ハンズオンでは計算の検算で振り分ける形にしました。
-- 明細の合計＝小計、小計＋消費税＝合計 のどちらかが合わなければ、人が確認する
NOT (coalesce(items_sum = subtotal AND subtotal + tax = total, false)) AS needs_review
実行結果はこうなりました。min_confidence は、明細以外の項目の信頼度のうちいちばん低い値です。
| case_id | items_sum | subtotal | tax | total | min_confidence | needs_review |
|---|---|---|---|---|---|---|
| case_01 | 166000 | 166000 | 16600 | 182600 | 0.2 | false |
| case_02 | 88000 | 88000 | 8800 | 96800 | 1 | false |
| case_03 | 68000 | 68000 | 6800 | 74800 | 0.05 | false |
| case_04 | 442000 | 442000 | 44200 | 486200 | 0.4 | false |
| case_05 | 117000 | 118000 | 11700 | 128700 | 1 | true |
case_05は手書きの見積書をスキャンした画像です。
明細を足すと48,000＋16,000＋42,000＋4,000＋7,000＝117,000円ですが、書かれた小計は118,000円で、合いません。これは書いた人の計算の誤りとして仕込んだもので、ai_extract は書かれた値を正しく読んでいます。このときの信頼度は1で、信頼度を見てもこの誤りには気づけません。一方で、計算はすべて合っているcase_03の信頼度は0.05でした。
信頼度はモデルが読み取りにどれだけ確信を持っているかの指標で、書類の中身が正しいかを判定するものではありません。当然といえば当然ですが、実際に並べてみると、信頼度だけで人の確認を振り分けると取りこぼしと過剰な差し戻しの両方が起きうることが分かります。また、信頼度の値は実行のたびに揺れます。人が確認する書類は、業務のルール(ここでは検算)で振り分けるのが確実です。
抽出の品質そのものを測りたいときは、付録の appendix_evaluation で正解データと突き合わせます。スキーマの説明やモデルを変えたときに、同じ正解データで比べれば良くなったか悪くなったかが数字で分かります。
つなぐ: 事故1件＝1行のGoldテーブル
05_gold_join では、3つのSilverを case_id でつなぎます。この工程はLLMを呼ばず、SQLの結合と条件式だけです。
食い違いの定義は業務で決めるルールで、ハンズオンでは次の3つにしました。
| 食い違い | 定義 |
|---|---|
| 損傷位置の食い違い | 通話での申告・写真・見積書のあいだで、損傷の位置が一致しない |
| 申告の程度と金額の食い違い | 通話で「軽微」と申告しているのに、見積の合計が20万円以上 |
| 見積書の検算エラー | 04の検算が合わなかった |
filter(array(
CASE WHEN size(array_distinct(filter(array(a.damage_position, i.damage_position, f.damage_position), x -> x IS NOT NULL))) > 1
THEN '損傷位置の食い違い' END,
CASE WHEN a.damage_severity = '軽微' AND f.total >= 200000
THEN '申告の程度と金額の食い違い' END,
CASE WHEN f.needs_review THEN '見積書の検算エラー' END
), x -> x IS NOT NULL) AS findings
損傷の位置は3つとも同じ選択肢から取り出しているので、文字列の比較だけで済みます。02で語彙をそろえておいたのが、ここで効いてきます。
案件の一覧はBronzeから作り、そこに3つのSilverを LEFT JOIN しています。どれかの形式が届いていなくても、案件の行は残ります。
1回目の5件では、3件に食い違いが見つかりました。
| case_id | findings |
|---|---|
| case_03 | 損傷位置の食い違い |
| case_04 | 申告の程度と金額の食い違い |
| case_05 | 見積書の検算エラー |
case_03は冒頭で写真と見積書を見た案件で、申告の「左前」に対して写真と見積書が「右後ろ」になっています。case_05は、前の節で見た手書きの見積書です。
担当者が最初に見るべき一覧が、通話の要点(summary)付きで出てくる形です。
運用に乗る: 新しく届いたファイルだけを処理する
最後の 06_add_new_files で、新しい事故2件(case_06・07)のファイルを置いて01〜05をもう一度実行します。
各工程が「処理済み」をどう判断しているかは次のとおりです。
| 工程 | 処理済みの判断 |
|---|---|
| 01 Bronze | Auto Loaderが取り込んだファイルをチェックポイントに記録している |
| 02〜04 Silver | Bronzeの行のうち、Silverに同じ path の行が無いものだけを選んでいる(LEFT ANTI JOIN) |
| 05 Gold | LLMを呼ばないので、毎回すべてを作り直す |
Silverの増分処理は、Auto Loaderのチェックポイントではなく LEFT ANTI JOIN で書いています。LLMを呼ぶ工程の「処理済み」を、結果のテーブルそのものに持たせる形です。失敗した行も error 付きでSilverに入るので、そのままでは再処理されない点には注意してください。再処理したいときは、その行を消してから実行し直します。
Goldだけ増分にしていないのは、費用がほとんどかからないからです。LLMを呼ぶ工程だけ増分にして、安い工程は毎回作り直す。仕組みを単純に保てます。
06では dbutils.notebook.run で01〜05を順に呼びます。この順番は、ジョブにしたときのタスクの順番と同じです。
for nb in ["01_bronze_ingest", "02_silver_audio", "03_silver_image", "04_silver_form", "05_gold_join"]:
dbutils.notebook.run(f"./{nb}", 1800, {"catalog": CATALOG}) # 1800: 1本あたりの時間の上限(秒)
実行後、Silverの各行の処理時刻(processed_at)を工程ごとに数えると、1回目の5件と2回目の2件に分かれます。
| 工程 | 件数 | 案件 |
|---|---|---|
| 02 音声 | 5 | case_01〜05 |
| 02 音声 | 2 | case_06, case_07 |
| 03 写真 | 5 | case_01〜05 |
| 03 写真 | 2 | case_06, case_07 |
| 04 見積書 | 5 | case_01〜05 |
| 04 見積書 | 2 | case_06, case_07 |
2回目は、どの工程も新しい2件だけをモデルに渡しています。処理済みの5件にはもう一度課金されていません。
Goldは毎回作り直すので7件がそろいます。case_07は、通話では右前のドアと申告しているのに写真と見積書は左前のドア、という案件として作ってあり、06の最後のセルで findings に「損傷位置の食い違い」が入ることを確かめます。
ジョブで自動実行する
01〜05は、そのままジョブのタスクにできます(01 →〔02・03・04を並列〕→ 05)。landing/ にファイル到着トリガーを付ければ、ファイルが届くたびに自動で実行されます。ジョブから実行するときは、パラメータ catalog でカタログ名を渡せるようにしてあります。
ファイル到着トリガーが動くかは、Volumeがある外部ロケーションの設定(ファイルイベント)によって変わるので、事前に確かめてください。
やり直すとき
00_setup の RESET = True にして実行すると、作ったテーブル・置いたファイル・Auto Loaderのチェックポイントを消します。チェックポイントを消さないと、同じファイルを置き直しても処理済みとみなされて取り込まれません。
まとめ
ハンズオンを作って分かったことをまとめます。
- 音声・画像・帳票とも「Bronzeの未処理分を選ぶ → AIで取り出す → Silverに追記する」という同じ形で書ける。違うのは取り出す手段だけ
- 音声は基盤モデルAPIに直接入力、画像は
ai_query、帳票はai_parse_document→ai_extract。ai_queryのfilesで渡せるのは画像なので、音声はPythonから呼ぶ - 課金の形も手段で変わる。トークン課金、バッチ推論の時間課金、AI Functionsの課金
- 損傷の位置などの選択肢を3形式で共通にしておくと、つなぐ工程が文字列の比較だけで済む
-
ai_extractのスキーマの説明に書いた正規化(和暦→西暦、「様」を除く)は、そのまま結果に反映される -
ai_extractの信頼度は、書類の中身が正しいかを判定するものではない。人の確認は検算のような業務のルールで振り分ける - LLMを呼ぶ工程は
LEFT ANTI JOINで増分にし、安い工程は毎回作り直す。再実行しても処理済みのファイルに課金されない -
ai_parse_documentの環境要件は、最初のノートブックで確かめておくと途中で止まらない
一番の収穫は、検算で拾ったcase_05の信頼度が1だったことでした。AIが正しく読んだ結果と、書類の中身が正しいことは別の話です。取り出しをAIに任せるほど、その後ろに業務のルールでの確認を置く設計が効いてきます。
リポジトリはそのまま動くので、ぜひ手元のワークスペースで試してみてください。
参考リンク
- databricks-japan-bootcamp / unstructured-data-pipeline-workshop
- Auto Loader
- ai_query関数
- ai_queryによるバッチ推論
- ai_parse_document関数
- ai_extract関数
- 構造化出力
- 音声・動画モデルの照会
- パートナーが提供するAI機能
- サーバーレス環境の設定
- ファイル到着トリガー
- Gitフォルダ