はじめに
Databricks AI/BI Genie（旧 Genie Space）の Benchmark 機能は、テスト質問セットに対する Genie の応答精度を定量評価できます。しかし、UI 上の詳細比較（生成 SQL と正解 SQL の差分）は実行から約1週間で expire し、閲覧できなくなります。
「先週の結果と比較したい」「改善前後の推移を可視化したい」といったニーズには、UI だけでは対応できません。
本記事では、Databricks SDK for Python の eval 系 API を使い、ベンチマーク結果をプログラムで取得し、Unity Catalog テーブルに永続保存する方法を解説します。Free Edition で検証済みのコードを掲載しています。
前提条件
- Databricks ワークスペース（Free Edition 可）
- Genie Space でベンチマーク実行済み（少なくとも1回の Run が存在すること）
-
databricks-sdk0.121.0 以上（eval 系メソッドが含まれるバージョン）
UI の制約：詳細比較が1週間で消える
Benchmark 実行後、UI の Evaluation タブでは各質問の正誤・生成 SQL・正解 SQL を確認できます。しかし以下の制約があります。
- 実行から約1週間経過すると、個別質問の詳細比較が「Details expired」となり閲覧不可になります
- Run 一覧（正答率のサマリ）は残りますが、「どの SQL が生成されたか」の詳細は消えます
-
information_schemaやシステムテーブルにはベンチマーク結果は格納されていません
つまり、改善履歴を追跡するには、結果を自分で取得して保存する仕組みが必要です。
解決策：eval 系 API による3段階取得
Databricks SDK には Genie のベンチマーク結果を取得する eval 系メソッドが3つ用意されています。
| メソッド | 取得内容 |
|---|---|
genie_list_eval_runs |
実行一覧（正答率サマリ） |
genie_list_eval_results |
各質問の正誤ステータス・正解SQL |
genie_get_eval_result_details |
生成SQLとその実行結果 |
この3つを順に呼ぶことで、UI で見える情報（＋ UI が隠す内部情報）を全て取得できます。
実装
Step 1: SDK のインストール
%pip install -U databricks-sdk
dbutils.library.restartPython()
バージョン 0.121.0 以上が必要です。pip show databricks-sdk で確認してください。
Step 2: クライアント初期化
from databricks.sdk import WorkspaceClient
w = WorkspaceClient()
Databricks ノートブック内では WorkspaceClient() が自動的に環境の認証情報を解決します。引数不要です。
Step 3: Eval Runs の取得
SPACE_ID = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # 対象スペースのID
resp = w.genie.genie_list_eval_runs(space_id=SPACE_ID)
for run in resp.eval_runs:
print(run.as_dict())
レスポンス構造：
{
"eval_runs": [
{
"created_timestamp": 1752826598551,
"eval_run_id": "...",
"eval_run_status": "DONE",
"last_updated_timestamp": 1752826645262,
"num_correct": 4,
"num_needs_review": 0,
"num_questions": 4,
"run_by_user": 123456789012345
}
]
}
注意点として、戻り値は GenieListEvalRunsResponse オブジェクトであり、直接イテレーションできません。.eval_runs プロパティ経由でリストにアクセスします。list(w.genie.genie_list_eval_runs(...)) は TypeError になります。
Step 4: Eval Results の取得
import json
run_id = resp.eval_runs[0].eval_run_id
results = w.genie.genie_list_eval_results(
space_id=SPACE_ID,
eval_run_id=run_id
)
for r in results.eval_results:
print(r.question, "→", r.status)
レスポンス構造：
{
"eval_results": [
{
"benchmark_answer": "SELECT ... FROM ...",
"benchmark_question_id": "...",
"created_by_user": 123456789012345,
"question": "What are the top selling products?",
"result_id": "...",
"space_id": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"status": "DONE"
}
]
}
benchmark_answer フィールドに正解 SQL（Ground truth SQL）が入っています。これは UI の Benchmark 設定で登録した SQL そのものです。
Step 5: Eval Result Details の取得
detail = w.genie.genie_get_eval_result_details(
space_id=SPACE_ID,
eval_run_id=run_id,
result_id=results.eval_results[0].result_id
)
print(json.dumps(detail.as_dict(), indent=2))
レスポンス構造：
{
"actual_response": [
{
"response": "SELECT p.product_name, SUM(s.quantity) ... FROM ...",
"response_type": "SQL",
"sql_execution_result": {
"manifest": {
"chunks": [{"chunk_index": 0, "row_count": 10}],
"schema": {
"column_count": 2,
"columns": [
{"name": "product_name", "type_name": "STRING"},
{"name": "total_quantity", "type_name": "LONG"}
]
}
}
}
}
]
}
actual_response[0].response に Genie が生成した SQL が入っています。これが UI の「Generated SQL」に相当し、expire 後に見られなくなる情報です。
完成版ノートブック
上記の3ステップを統合し、関数化したノートブックです。2つのセルで「API 取得 → Unity Catalog 保存」が完結します。
Cell 1: ベンチマーク結果の取得
from databricks.sdk import WorkspaceClient
from datetime import datetime
def collect_benchmark_results(w, space_id):
"""指定スペースの全 Run × 全質問のベンチマーク結果を取得する"""
records = []
runs_resp = w.genie.genie_list_eval_runs(space_id=space_id)
if not runs_resp.eval_runs:
return records
for run in runs_resp.eval_runs:
run_id = run.eval_run_id
results_resp = w.genie.genie_list_eval_results(
space_id=space_id, eval_run_id=run_id
)
for result in results_resp.eval_results:
detail = w.genie.genie_get_eval_result_details(
space_id=space_id,
eval_run_id=run_id,
result_id=result.result_id,
)
generated_sql = ""
if detail.actual_response:
generated_sql = detail.actual_response[0].response or ""
records.append({
"space_id": space_id,
"run_id": run_id,
"run_timestamp": datetime.fromtimestamp(
run.created_timestamp / 1000
).isoformat(),
"run_status": str(run.eval_run_status),
"num_correct": run.num_correct,
"num_questions": run.num_questions,
"question": result.question,
"status": str(result.status),
"ground_truth_sql": result.benchmark_answer,
"generated_sql": generated_sql,
"result_id": result.result_id,
})
return records
w = WorkspaceClient()
SPACE_ID = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
all_records = collect_benchmark_results(w, SPACE_ID)
print(f"取得完了: {len(all_records)} 件")
Cell 2: Unity Catalog テーブルへの保存
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from datetime import datetime as dt
SCHEMA_DEF = StructType([
StructField("space_id", StringType(), False),
StructField("run_id", StringType(), False),
StructField("run_timestamp", StringType(), False),
StructField("run_status", StringType(), True),
StructField("num_correct", IntegerType(), True),
StructField("num_questions", IntegerType(), True),
StructField("question", StringType(), False),
StructField("status", StringType(), True),
StructField("ground_truth_sql", StringType(), True),
StructField("generated_sql", StringType(), True),
StructField("result_id", StringType(), False),
StructField("collected_at", StringType(), False),
])
def save_to_table(spark, records, full_table):
"""初回はテーブル作成、2回目以降は MERGE で重複排除しながら追記する"""
for rec in records:
rec["collected_at"] = dt.now().isoformat()
df = spark.createDataFrame(records, schema=SCHEMA_DEF)
if not spark.catalog.tableExists(full_table):
df.write.saveAsTable(full_table)
print(f"テーブル作成完了: {full_table}")
else:
df.createOrReplaceTempView("new_results")
spark.sql(f"""
MERGE INTO {full_table} AS target
USING new_results AS source
ON target.run_id = source.run_id
AND target.question = source.question
AND target.space_id = source.space_id
WHEN NOT MATCHED THEN INSERT *
""")
print(f"MERGE完了: {full_table}")
CATALOG = "dev" # ワークスペースの既存カタログを指定
SCHEMA = "genie_benchmarks"
TABLE = "eval_results"
FULL_TABLE = f"{CATALOG}.{SCHEMA}.{TABLE}"
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CATALOG}.{SCHEMA}")
save_to_table(spark, all_records, FULL_TABLE)
display(spark.sql(f"SELECT * FROM {FULL_TABLE}"))
MERGE の ON 条件（run_id + question + space_id）で同一質問の重複挿入を防ぐため、定期ジョブで繰り返し実行しても安全です。
ワークスペース横断版
複数ワークスペースの Genie Space を一括で取得する場合は、WorkspaceClient に接続先を明示指定します。Cell 1 の collect_benchmark_results 関数をそのまま再利用できます。
WORKSPACES = [
{
"host": "https://dbc-xxxxxxxxxx-xxxx.cloud.databricks.com",
"token": dbutils.secrets.get(scope="genie", key="ws1_token"),
"space_ids": ["xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"],
},
{
"host": "https://adb-xxxxxxxxxxxx.azuredatabricks.net",
"token": dbutils.secrets.get(scope="genie", key="ws2_token"),
"space_ids": ["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"],
},
]
all_records = []
for ws_config in WORKSPACES:
w = WorkspaceClient(
host=ws_config["host"],
token=ws_config["token"],
)
for space_id in ws_config["space_ids"]:
all_records.extend(collect_benchmark_results(w, space_id))
print(f"全ワークスペース合計: {len(all_records)} 件")
トークンは dbutils.secrets から取得しています。Secret Scope の設定方法は 公式ドキュメント を参照してください。
ジョブ化による定期収集
ノートブックを Databricks ジョブとしてスケジュールすれば、expire 前に自動収集できます。
# ジョブ設定の推奨値
# - スケジュール: 毎日1回（cron: 0 0 9 * * ?）
# - クラスタ: Serverless
# - リトライ: 1回
# - アラート: 失敗時にメール通知
1日1回の実行で十分です。ベンチマーク結果は最大1週間保持されるため、日次収集で取りこぼしは発生しません。
まとめ
| 課題 | 解決策 |
|---|---|
| UI の詳細比較が1週間で消える | eval API で取得して保存 |
information_schema に無い |
SDK の genie.genie_* メソッドで取得 |
| 複数ワークスペースを横断したい |
WorkspaceClient(host, token) で接続先を切替 |
| 履歴を蓄積したい | Unity Catalog Delta テーブルに MERGE |
Genie Space のベンチマーク機能は改善サイクルを回すための重要なツールですが、結果の永続化は自前で実装する必要があります。本記事のコードを定期ジョブ化することで、改善履歴の可視化や回帰検知に活用できます。
関連記事
補足：API の注意点
- eval 系 API は Beta（2026年7月時点）です。2026年3月19日に Beta として公開されて以降、GA 昇格のアナウンスはありません。メソッド名やレスポンス構造が変更される可能性があります
- レスポンスオブジェクトは直接イテレーションできません（
list()で囲むとTypeError）。必ず.eval_runs/.eval_resultsプロパティ経由でアクセスしてください -
genie_get_eval_result_detailsは1質問ずつしか取得できないため、質問数が多い場合はループが必要です -
genie_list_eval_runs/genie_list_eval_resultsにはpage_size/page_tokenパラメータがあります。質問数が多い場合はページネーションが必要です -
genie_create_eval_run(space_id, benchmark_question_ids)でベンチマーク実行自体もプログラムからトリガーできます -
run.eval_run_statusやresult.statusは Python の enum オブジェクト（EvaluationStatusType）です。json.dumpsでシリアライズする場合はstr()で文字列に変換してください - Free Edition の Serverless SQL Warehouse でも動作確認済みです