Oracle PL/SQL→Databricks マイグレーションで作った LLMコンバージョンツールと dbt 評価基盤の紹介
Databricks Advent Calendar 2025 シリーズ2 25日目の記事です。Oracle を Databricks にマイグレーションする機会があり、LLM を用いた自動コンバージョンツールとコンバージョン結果を評価するための dbtの評価基盤を構築しました。本記事では、コンバージョン評価プロセス全体の流れとその開発・検証の過程で得られた技術的な Tips をご紹介します。LLMを用いたマイグレーション全般に適用できる内容として整理しているため、同じような悩みを持つ方の参考になれば幸いです。
コンバージョン評価プロセス全体の流れ
最初にコンバージョン評価プロセス全体の流れを紹介します。PL/SQL のコンバージョンには LLM ベースのコンバージョンツールを使用しています。また、コンバージョン結果をデータ比較し、評価する仕組みも必要です。評価基盤にはdbtを利用しています。LLMコンバージョンツールやdbt 評価基盤の詳細は次の章で説明し、まずは全体のプロセスの流れを説明します。
データ評価を行うためには、正解データを用意する必要があります。そこで、既存の Oracle 環境において日次で実行されている Procedure バッチ処理完了後のテーブルを、Databricks にロードしています。ロードしたテーブルには日付ごとにサフィックスを付与し、カタログを分けて管理しています（例：prod_oracle_20251225）。日付ごとに断面を分けている理由は、静的なデータ断面を用いることで、複数のデータ評価者がそれぞれ割り当てられた日付断面を使用し、互いに影響を受けることなく、独立した環境でデータ評価を行えるようにするためです。また、この後紹介するProcedureのデータ比較評価に必要な環境を作るためにも利用されます。
ロードされたOracleのデータを用いて、Procedureのデータ比較評価に必要な環境(以降、評価環境と呼びます)を構築します。
評価環境で使用するデータは、次の 2 種類です。
1 つ目は、日次で連携され、Procedure のインプットとなるテーブルです（以降「インプットテーブル」と呼びます）。
2 つ目は、インプットテーブル以外で、Procedure の実行によって更新されるテーブルです。
評価者はそれぞれデータ比較する対象日をdbt の設定ファイル内で指定して評価を行います。インプットテーブルには、指定された日付断面の Oracle テーブルを使用し、それ以外のテーブルについては、指定日付の前日断面の Oracle テーブルを利用します。このように構成することで、Procedure 実行後の評価環境は、テーブルが更新され、Oracle におけるバッチ処理完了後の状態と一致します。
また、評価環境の構築時には、データレベルの前処理が必要です。Oracle と Databricks ではデータ型の挙動に差異があるためです（例：CHAR 型は Oracle ではCHARサイズに応じてスペース埋めされますが、Databricks ではされません）。こうした差異は、 Procedure のデータ比較評価結果をもとに、前処理として吸収しています。次にLLMコンバージョンツールを使い、Oracle PL/SQLをDatabricks Procedureにコンバージョンします。Databricks Procedureについては以下に詳細を書きました。
LLMコンバージョンツールでは正規表現等の「ルール処理」と「LLM処理」を組み合わせて自動コンバージョンしています。Databricks ProcedureはSQLウェアハウスで実行できるので、変換後にAPIを使い、「CREATE PROCEDURE」でUnity Catalogに登録します。Syntaxレベルのエラーがあった場合、「CREATE PROCEDURE」実行時に検知することができます。PySparkでのジョブ起動等ないため、改善サイクルを高速に回すことができました。Syntaxエラーとなった場合、エラーメッセージとコンバージョン結果をプロンプトに組み込み、リトライします。数回リトライしても改善しなかった場合、「人間」が確認し、LLMコンバージョンツールに修正等フィードバックします。Syntaxレベルのエラーが改善されるまで、繰り返します。
次にdbtを用いて、Procedureの実行とデータ比較評価を行います。dbt Pythonモデルを用いてProcedureを実行すると、先ほど紹介した評価環境のテーブルが更新されます。更新された結果はOracleからロードした当日断面と同じ結果になっているため、データ比較することで、差分の確認ができます。
データ比較の差分結果を「人間」が確認し、どのタスクにフィードバックするか判断し修正することで、改善サイクルを回します。データ評価のフィードバック先は多岐に渡ります。依存関係等のメタ情報の不備の場合もあれば、テーブル定義、データの前処理、LLMコンバージョンツール、コンバージョン結果、UDF等考えられます。データの差分結果を考慮して、適切なタスクにフィードバックして、改善します。
Procedure の評価が完了した後は、日次で連携されてくるProcedureのインプットテーブルの評価を行います。インプットテーブルはオブジェクトストレージ(S3等)経由で連携を想定します。連携時にはまだProcedureのフィードバックを反映したインプットテーブルになっていないため、前処理を行い、評価環境のインプットテーブルと一致するようにデータ比較します。
全体の評価フローは、まず Procedure 単体テストから始めます。次に Procedure 同士の依存関係を考慮した結合テストを行い、その後、Procedureのテスト結果が反映された評価環境を正解データとし、インプットテーブルの単体テストをします。最後に、インプットテーブルと Procedure を組み合わせた結合テストを行います。一連の評価が完了した後は、日次バッチとして実行し、継続的に監視します。並行稼働期間中は、データの差分や不整合を継続的に検知できるようになります。
以上が、コンバージョン評価プロセス全体の流れです。次章からは具体的なLLMコンバージョンツールとdbt評価基盤の技術的なTipsをご紹介していきます。
LLMコンバージョンツール
LLMコンバージョンツールで行っていることの詳細をご紹介します。
ルール処理(前処理・後処理)
LLM は非決定的であり、コンテキストウィンドウや出力トークンにも制限があります。そのため、正規表現などで対応可能な部分についてはルールベース処理を採用し、正規表現では対応が難しい箇所のみ LLM を利用しています。ルールベースで対応可能な部分はルール処理に任せることで、変換精度と安定性を高めています。
コメント削除等によるコンテキスト削減
前処理では、 PL/SQL に含まれるコメントや不要な改行を削除し、LLM に渡す前のコンテキスト量を削減しています。これにより、コンテキスト量をおおよそ 1/3 程度削減することができました。空白等も当初は削除していましたが、Procedureのデータ比較を通して、削除してはいけないことがわかったので、対象から外しています「normalize_whitespace: false」。
input_preprocessing: remove_comments: true # Remove all SQL comments (-- and /* */) from input remove_empty_lines: true # Remove empty lines from input normalize_whitespace: false # Normalize multiple spaces/tabs to single space compress_sql_formatting: true # Remove line breaks within SQL statements for compact formatting
正規表現での変換
正規表現で変換可能な箇所については、すべて正規表現による「ルール処理」で対応しました。具体的には、日付フォーマットの変換や日本語変数をバッククオートで囲む処理、Oracle 互換の関数変換などを正規表現で変換しました。LLM は変換結果が不安定になるケースもあるため、正規表現で対応できる部分については、ルールベース処理の方がより正確かつ安定した変換が可能です。
- （例）日本語を含む変数に対するバッククオートの付与
日本語を含む変数名に対するバッククオートの付与は、LLM では特に変換漏れが発生しやすいポイントでした。英語と日本語が混在するケースでは、LLM がカラムを見落とすことが多く、安定した変換が困難でした。
INSERT INTO `TEMP_利用者`( ID, `名前` ) SELECT USER.ID USER.`ユーザー` FROM USER
- （例）日付フォーマット変換
PL/SQL と Databricks では日付フォーマットの指定方法に差異があるため、これらは正規表現によるルール処理で変換しています。
YYYYMMDD → yyyyMMdd yyyy/mm/dd → yyyy/MM/dd YYYY/MM/DD → yyyy/MM/dd YYYY/mm/dd → yyyy/MM/dd
- （例）Oracle 互換関数の変換
Oracle 互換の関数については、Databricks 側に互換 UDF を用意して利用しています。関数の変換等はLLMよりもルールベースの方が正確かつ、LLMで消費するコンテキスト量も減らすことができます。
function_mappings: TO_DATE: "oracle_compatible.TO_DATE" # TO_DATE(str, fmt) → oracle_compatible.TO_DATE(str, fmt) TO_CHAR: "oracle_compatible.TO_CHAR" # TO_CHAR(value, fmt) → oracle_compatible.TO_CHAR(value, fmt) ADD_MONTHS: "add_months" # ADD_MONTHS(date, n) → add_months(date, n) NVL: "coalesce" # NVL(expr1, expr2) → coalesce(expr1, expr2) TRIM: "oracle_compatible.TRIM" # TRIM(col) → oracle_compatible.TRIM(col) RTRIM: "oracle_compatible.RTRIM" # RTRIM(col) → oracle_compatible.RTRIM(col) LTRIM: "oracle_compatible.LTRIM" # LTRIM(col) → oracle_compatible.LTRIM(col) SUBSTRB: "SUBSTR" SYSTIMESTAMP: "current_timestamp()" # SYSTIMESTAMP → current_timestamp() SYSDATE: "current_timestamp()" # SYSDATE → current_timestamp() TO_NUMBER: "oracle_compatible.TO_NUMBER" # TO_NUMBER(col) → oracle_compatible.TO_NUMBER(col) TRUNC: "FLOOR" # TRUNC(number, scale) → FLOOR(number, scale) current_date: "current_timestamp()" ....
LLM処理
ルール処理だけでは対応が難しい部分は「LLM処理」で対応しています。行数の多いPL/SQLをコンバージョンするための工夫やLLMの不安定さを取り除くための方法を紹介します。
数千行以上のPL/SQL をコンバージョンする方法：会話履歴・出力の分割と結合・リクエストパラメータ制御
数千行を超えるような行数の多い PL/SQL は、前処理によってコンテキスト量を削減しても、モデルの出力トークン制限を回避することは難しいです。
例えば Claude 3.7 Sonnet では、コンテキストウィンドウ(200,000トークン)および出力トークン(64,000トークン)に上限があり、そのままリクエストすると出力トークン制限に達してしまいます。また、モデルの許容するコンテキスト上限ギリギリでリクエストした場合、出力の一部が途中で欠落することも確認できました。
そこで、リクエスト時に余裕を持った出力トークン 「max_output_tokens 」を設定し、出力トークンの上限に達した場合は、途中までの変換結果を会話履歴として保持し、続きから再生成する仕組みを実装しました。この方法により、ベンダーが提供するモデルの出力トークン制限を回避しています。
具体的には、以下の流れで処理を行っています。
conversion_settings: default_temperature: 0.0 default_max_tokens: 60000 # Set max_output_tokens with a sufficient margin. default_stream: true # Enable streaming output by default
出力トークンの上限に達した場合は、生成済みの結果を会話履歴として保存します。finish_reason が「length」の場合、次の LLM リクエストではその履歴をプロンプトとあわせて渡し、前回の続きからコードを再生成します。この処理を繰り返し、分割生成された結果を CONCAT して統合することで、行数の多い Procedure の変換を行っています。
# Continue if truncated due to length limit while final_finish_reason == 'length' and continuation_count < max_continuations: continuation_count += 1 print(f"{context_prefix}Response truncated, requesting continuation {continuation_count}/{max_continuations}...") # Add continuation request to conversation history current_messages.append({"role": "user", "content": continuation_prompt or "Please continue from where you left off. Provide only the continuation without repeating what's already provided."}) try: if stream: # Use streaming version for continuation continuation_content, finish_reason, cont_token_usage = self._single_stream_request_with_metadata( current_messages, temperature, max_tokens, context_prefix ) except Exception as e: ...... full_text += continuation_content current_messages.append({"role": "assistant", "content": continuation_content}) final_finish_reason = finish_reason
PL/SQL のコンバージョンでは正確性と一貫性が重要となるため、「temperature」は低めに設定し、出力の揺らぎを抑えています。また、行数の多いファイルでは タイムアウトが発生するするため、「stream」オプションを利用してストリーミングで結果を受け取ることで回避できます。さらに、出力トークンは意図的に小さく設定し、上限に達した場合は会話履歴を用いた分割生成を行うことで、出力トークン制限に対応しています。出力トークンを意図的に小さく設定しないと、LLM側で勝手に出力結果を削ったり、サマライズしたりしてしまうので、注意が必要です。
以下のサンプルはOpenAI Client を使用し、ストリーム処理で、Databricksのモデルサービングを利用する例です。
Use foundation models | Databricks on AWS
def chat_completion_stream(self, messages: List[Dict[str, str]], temperature: Optional[float] = None, max_tokens: Optional[int] = None) -> Iterator[Dict[str, Any]]: """Send streaming chat completion request to Databricks Model using OpenAI SDK""" params = { 'model': self.model_id, 'messages': messages, 'temperature': temperature } if max_tokens is not None: params['max_tokens'] = max_tokens # Use OpenAI SDK for streaming (following working sample.py approach) with self.openai_client.chat.completions.stream(**params) as stream: for event in stream: if event.type == "chunk": print(f'raw chunk finish_reason: {event.chunk.choices[0].finish_reason}') finish_reason = event.chunk.choices[0].finish_reason if event.type == "content.delta": # event.delta is the text fragment (plain string) chunk_text = event.delta print("content.delta") print(event) # Create OpenAI-compatible chunk format chunk = { 'choices': [{ 'delta': { 'content': chunk_text }, 'index': 0 }] } yield chunk elif event.type == "error": # Return error in expected format yield {"error": event.error} break elif event.type == "content.done": # Debug: check event attributes print(f"🔍 content.done event: {event}") # Send final chunk with actual finish_reason final_chunk = { 'choices': [{ 'delta': {}, 'index': 0, 'finish_reason': finish_reason }] } yield final_chunk break
このようにして、LLM モデルの出力トークン制限を回避しつつ、コンテキストウィンドウの上限近くまで対応できるようになります。
ただし、この処理をしてもコンテキストウィンドウの制限を回避することはできないので、数万行レベルのPL/SQLは事前に分割しておく必要があります。また、5,000 行を超える PL/SQLでは変換精度が低下する傾向があるため(コンテキストロット問題：入力トークン量(会話履歴)増加に伴う能力低下現象)、一定の行数を超える場合は、前処理として PL/SQL自体を事前に分割した方が良さそうです。会話履歴を活用した分割変換は、精度を維持したまま制限を回避できる有効な手法でした。一方で、履歴情報の渡し方ではコンテキスト圧縮なども試しましたが、十分な効果は得られませんでした。PL/SQL のコメントはできれば残したいと思いますが、精度とトレードオフとなってくるので、コンテキストウィンドウの制限がある現時点ではバランスを見て判断した方がいいと思います。
プロンプトの継続的改善とバージョン管理
PL/SQL を Databricks Procedure に変換するにあたり、LLM が十分な知識を持っていなかったため、公式ドキュメントやベンダー側のベストプラクティスをもとにベースプロンプトを作成しました。ただし、ドキュメントに記載のない仕様も多くあるため、データ評価結果をプロンプトにフィードバックし、継続的に改善しました。当初は関数変換も LLM に任せていましたが、不安定さや変換漏れが多く、また評価結果を反映しすぎるとコンテキスト肥大によるデグレも発生しました。そのため最終的には、「ルール処理」で対応できない部分のみに 「LLM 処理」を限定し、コンテキストを抑えることで精度と安定性を両立しています。また、LLMが十分な知識を持っていないため、「LLM処理」もしくは「ルール処理」どちらにも変換方法を明示していないものは不安定な出力となりました。コンバージョンでは変換結果の一貫性が重要となるため、基本的にはどちらかには変換方法を明示しています。
また、プロンプトのバージョン管理は GitHub のみで行っています。改善サイクルを高速に回すことを重視していたため、ローカル環境から実行でき、VS Code エディタ内で完結する形で GitHub 管理としました。デグレが発生した場合も、プロンプトファイルのコミットを戻すだけで対応できます。
Syntaxエラー自動リトライによる出力の安定化
PL/SQL コンバージョンの単体テストでは、Syntax・Runtime・データレベルの 3 段階で評価しています。このうち Syntax レベルの評価は、LLM コンバージョンツール内で、自動修正しています。Runtime・データレベルはデータ起因や他の要因も考えられるため、LLM コンバージョンツール内では対応していないです。
「コンバージョン評価プロセス全体の流れ」で紹介した通り、コンバージョン結果をAPI 経由で SQL ウェアハウスに対して 「CREATE PROCEDURE」 を実行し、Syntax エラーがあれば即時検知されます。LLM の不安定さを考慮し、プロンプトにコンバージョン結果とSyntax エラーメッセージを加え、最大 3 回までリトライし、それでも解消しないものは「人間」が確認してツールにフィードバックしています。ツール内の「LLM処理」を改善するか、「ルール処理」を改善するかはエラー内容を見て判断しました。エラー内容やSyntaxエラーとなったコンバージョン結果はファイルに出力し、トレーシング可能な形で管理し、フィードバックの参考にしています。先ほどもお伝えしたとおり、改善サイクルを高速に回すため、ローカル環境で完結できることを重視していました。そのため、コンバージョン結果やSyntaxエラー内容をファイルとして出力し、トレーシングできる運用を取りました。
以下はSyntaxエラーリトライ時のサンプルとなります。プロンプトにコンバージョン結果とエラーメッセージを加えて、リトライしてるのが確認できます。
for attempt in range(1, max_retries + 1): print(f"{error_type} fix attempt {attempt}/{max_retries}") try: # Create fix prompt based on error type if error_type == "syntax_error": fix_prompt = f"""The following Databricks SQL has a syntax error. Please fix it. Error message: {current_error} SQL: {current_sql} Please return only the corrected SQL (no comments or explanations needed)."""
評価・改善はすべてローカル環境で高速に回し、全ProcedureのSyntaxレベルの改善は1週間程度で対応できました。Databricks Procedureがサポートされ、直接SQLウェアハウス実行できるようになったことも改善サイクルの高速化につながりました。
dbt 評価基盤
dbt 評価基盤で行っていることの詳細をご紹介します。
正解データの準備
「コンバージョン評価プロセス全体の流れ」で紹介した通り、日次で Oracle のテーブルを Databricks にロードしています。ロードには dbt Pythonモデルを利用しています。dbt Pythonモデルは Python で ETL ロジックを記述でき、Databricks の Notebook Job として実行可能です。SQL では難しい処理にも対応でき、SQLモデルとの依存関係も制御できます。dbtのエコシステムを利用できるので便利です。
Databricks configurations | dbt Developer Hub
Python models | dbt Developer Hub
以下のサンプルはdbt Pythonモデル内で、JDBCドライバを使い、Oracleのテーブルをロードするサンプルです。テーブル数が多い場合、テーブルごとにモデルを作成するとモデル数が膨大になってしまうので、マルチスレッド処理等で対応した方が良いと思います。Pythonが使えるので、柔軟な処理が可能です。
(例) Oracleに永続化されているテーブルをDatabricksへロード
import boto3 def model(dbt, session): dbt.config( materialized='table', tags=['sample'], ) host = dbt.config.get('dbx_host') ssm = boto3.client('ssm', region_name=dbt.config.get('region')) user = ssm.get_parameter(Name=dbt.config.get('jdbc_user_path'), WithDecryption=True)['Parameter']['Value'] pwd = ssm.get_parameter(Name=dbt.config.get('jdbc_password_path'), WithDecryption=True)['Parameter']['Value'] jdbc_url = ssm.get_parameter(Name=dbt.config.get('jdbc_url_path'), WithDecryption=True)['Parameter']['Value'] fetchsize = dbt.config.get('fetchsize') src = 'sample' print(f'table: {src}') data_q = f"(SELECT * FROM {src})" try: df = session.read.format('jdbc') \ .option('url', jdbc_url) \ .option('dbtable', data_q) \ .option('user', user) \ .option('password', pwd) \ .option('driver', 'oracle.jdbc.OracleDriver') \ .option('fetchsize', fetchsize) \ .load() except Exception as e: print(f"Error processing {src}: {e}") raise e return df
また、PL/SQLにも大きく２種類あり、バッチ処理として実行されているものと、レポーティングツールなどからテーブル関数のように利用され、最終的なSELECT 結果を取得するものがあります。前者は Oracle 上にテーブルとして永続化されていますが、後者は Oracle 上に結果が永続化されないため、そのままでは正解データを用意できません。そこで後者については、Databricks から PL/SQL を実行し、その実行結果を Databricks 上に永続化することで、正解データを準備できます。正解データの準備は、いずれの場合も dbt Python モデルを利用しています。
(例) PL/SQLのSELECT結果をDatabricksにロード
dbt Python モデルでは、jpype1==1.5.2 を利用して PL/SQL を実行し、実行結果を取得しました。
(当初は 「python-oracledb」 の利用を検討しましたが、一部のPL/SQLでライブラリのバグを踏んでしまい、タイムアウトする事象が発生しました。jpype1==1.5.3 は Issue があったため、安定して動作した 1.5.2 にバージョンを固定しています。PL/SQLが多いとバグを踏む機会も多くありました)
def execute_procedure(conn, spec: Dict[str, Any], fetch_full: bool = True, preview_rows: int = 10) -> Dict[str, Any]: proc_schema = spec.get('schema') proc_name = spec['name'] params_meta: List[Dict[str, Any]] = spec.get('parameters_full', []) call_sql = '{' + 'call ' + proc_schema + '.' + proc_name + '(' + ','.join(['?'] * len(params_meta)) + ')}' print(f"Executing: {call_sql}") # Force session date format to avoid ORA-01861 for procedures using TO_DATE without format mask try: fmt_stmt = conn.createStatement() fmt_stmt.execute("ALTER SESSION SET NLS_DATE_FORMAT='YYYY/MM/DD'") fmt_stmt.execute("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY/MM/DD HH24:MI:SS'") fmt_stmt.execute("ALTER SESSION SET TIME_ZONE='Asia/Tokyo'") fmt_stmt.close() except Exception as e: print(f"[WARN] ALTER SESSION failed: {e}") stmt = conn.prepareCall(call_sql)
評価環境の準備
「コンバージョン評価プロセス全体の流れ」で紹介した通り、Procedureの「評価環境」を構築します。先ほどお伝えした通り、評価環境を構築するためにOracleの当日断面(インプットテーブル)と前日断面を利用します。構築時にはProcedureのデータ比較評価で得たフィードバックを前処理に反映しています。
例えば CHAR 型の場合、Oracle では空文字が入った際に、定義された CHAR サイズ（例：CHAR(8)）に応じてスペース埋めされた値が格納されます。一方、Databricks では CHAR サイズによるスペース埋めは行われません。そのため、テーブルを作成する際に、デフォルト値としてスペース埋めされるような対応を入れています。同様に、CHAR 型以外でも Oracle の DATE 型は「時刻」も持っているため、DatabricksではDATE型ではなく、TIMESTAMP型として扱う必要があります。このようにテーブル定義を作成する際にOracleとDatabricksのデータレベルの差分を前処理で吸収する必要があります。これらは最初からわかっていたわけではなく、Procedure のデータ比較評価結果を進める中で、継続的に改善していきました。
以下はDatabricks上でテーブル定義の情報を取得するクエリです。JDBCドライバで Oracle から Databricks にロードしたテーブルのスキーマ情報を管理する「information_schema」と、Oracle でスキーマ情報を管理している 「dba_tab_columns 」を利用して、今回のマイグレーションで利用している テーブル定義を作成しています。CHAR型かつデフォルト値がない場合、デフォルト値としてCHARサイズに応じたスペース埋めをデフォルト値として設定しています。
with information_schema as ( SELECT table_schema, table_name, column_name, data_type, full_data_type, ordinal_position FROM {copy_src_database_day_before}.information_schema.columns WHERE lower(table_schema) IN ('schema_a', 'schema_b') ), dba_tab_columns as ( select distinct OWNER, table_name, column_name, case when trim(DATA_DEFAULT) in ('NULL','null') then Null when trim(DATA_DEFAULT) in ('SYSDATE', 'sysdate', 'systimestamp', 'current_timestamp', 'SYSTIMESTAMP') then 'CURRENT_TIMESTAMP()' when trim(DATA_DEFAULT) == "'1980/01/01'" then 'to_timestamp("1980-01-01","yyyy-MM-dd")' else trim(DATA_DEFAULT) end as DATA_DEFAULT, NULLABLE from {database}.{schema}.dba_tab_columns where lower(OWNER) IN ('schema_a', 'schema_b') ) select a.table_schema, a.table_name, a.column_name, a.data_type, a.full_data_type, CASE WHEN lower(a.data_type) = 'char' AND b.NULLABLE = 'N' AND trim(b.DATA_DEFAULT) = "' '" THEN CONCAT("'", REPEAT(' ', CAST(REGEXP_EXTRACT(a.full_data_type, '(?i)CHAR\\\\((\\\\d+)\\\\)', 1) AS INT)), "'") ELSE b.DATA_DEFAULT END as DATA_DEFAULT, b.NULLABLE from information_schema as a left outer join dba_tab_columns as b on lower(a.table_schema) = lower(b.OWNER) and lower(a.table_name) = lower(b.table_name) and lower(a.column_name) = lower(b.column_name) ORDER BY table_schema, table_name, ordinal_position asc
作成したテーブル定義に対して、Procedure のデータ評価結果を考慮した前処理を行ったうえで、データを INSERT しています。CHAR 型のデータについては、値がオールスペースの場合はそのまま扱い、それ以外の場合は RTRIM を適用しています。これは、データ比較を進める中で、WHERE 条件や GROUP BY においてスペースの有無が差分として検知されるケースがあったためです。
また、文字列系はかなりややこしく、CHARやSTRING、VARCHAR型で挙動が異なります。当初は評価環境を構築する際に、ORACLEからロード済みのテーブルをSELECTする際に、適切なデータ型にCASTし、CREATE TABLEして構築していました。しかし、文字列系はCASTしてもCASTしたデータ型になりませんでした(CHAR(8)でCASTしてもSTRINGになってしまったり)。そのため、まず空のテーブルを作ってからINSERTするようにしました。これらもProcedure のデータ比較評価結果を進める中で、継続的に改善していきました。CHAR ではなく STRING になってしまうと、Procedure 内の WHERE 条件などでスペースを指定している場合、スペースの扱いが異なり、条件にマッチせず、適切な処理結果にならなくなります。
if 'char' in data_type: # If column contains only whitespace characters, keep original value; otherwise apply rtrim and cast to full_data_type column_selects.append(f"CAST(CASE WHEN `{column_name}` RLIKE '^\\\\s+$' THEN `{column_name}` ELSE rtrim(`{column_name}`) END AS {full_data_type}) as `{column_name}`") elif 'timestamp' in data_type: column_selects.append(f"CAST(CASE WHEN `{column_name}` IS NULL THEN NULL ELSE `{column_name}` - INTERVAL 9 HOURS END AS {full_data_type}) as `{column_name}`") else: column_selects.append(f"`{column_name}`")
「コンバージョン評価プロセス全体の流れ」で紹介した通り、インプットテーブルは当日断面、それ以外は前日断面から取得しています。
# copy_src_database：prod_oracle_20251225 # copy_src_database_day_before：prod_oracle_20251224 if is_source_table: insert_query = f"INSERT INTO {target_database}.{copy_src} SELECT {columns_str} FROM {copy_src_database}.{copy_src};" else: insert_query = f"INSERT INTO {target_database}.{copy_src} SELECT {columns_str} FROM {copy_src_database_day_before}.{copy_src};"
評価環境は評価者ごとにローカル環境から割り当てられた日付断面を使い、構築しています。dbtの設定ファイル(dbt_project.yaml)に変数として日付情報をセットすることで、対象日の評価環境を構築しています。
Databricks Procedureの実行
「コンバージョン評価プロセス全体の流れ」で紹介した通り、Databricks Procedureはdbt Python モデルから実行しています。Databricks Procedureは新しい機能なので、不安はあったのですか、もし何か問題があればPython で処理できるため、Databricks Procedureの導入を決める上で、安心材料となりました。
いきなり全 Procedure をテストすると原因調査が困難になるため、dbtのタグ機能を活用し、Procedure 単位で実行を制御しています。「dbt tag」を利用することで、実行時に 特定のタグだけ実行・除外 することができます。この機能を利用することで、同じdbtモデルファイルを利用し、Procedureの単体テスト、依存関係を考慮したProcedureの結合テストが可能となります。
具体的には以下のような「dbt tag」を使っています。Procedrueの単体テスト、結合テストができるように「procedure名」、「procedureの属する親」、「procedureの属すスキーマ名」、「バッチ処理全体」単位と複数のタグを１つモデルに紐付け、依存関係も考慮した柔軟な実行単位の制御が可能です。
def model(dbt, session): dbt.config( tags=['<procedure_ parent_name>', '<procedure_name>', 'batch', '<procedure_schema>'] )
dbtのtagについて細かな挙動を確認してみた #dbt | DevelopersIO
また、「dbt.ref」を活用することで依存関係や同時実行数を考慮した並列実行が可能です。「dbt.ref」はdbtのSQLモデルとPythonモデル間でも定義することができます。マイグレーションをする場合、既存のPL/SQLの依存関係をそのまま持ってきたいことがあると思います。マイグレーションで依存関係を変更してしまうと、データ比較等で差分が出た場合、原因調査がむずかしくなるためです。Procedure間の依存関係は事前にメタ情報として定義し、メタ情報を利用して、dbt Pythonモデルファイルは自動生成しています。
def model(dbt, session): dbt.config( tags=['<procedure_ parent_name>', 'procedure_name', 'batch', '<procedure_schema>'] ) # Load reference models refs = [dbt.ref("procedure_a")]
About ref function | dbt Developer Hub
また、特定のカレンダー条件に応じて実行可否を制御したかったり、諸々の制約もあり、クラスターではなくSQL ウェアハウスのリソースを活用したいなどの特別な要件がある場合があります。そのような特殊な要件があっても、dbt Pythonモデルを使うことで柔軟に対応することができます。
通常の dbt は「1 モデル＝1 テーブル」を前提としていますが、Procedureの場合、「1モデル= 1Procedure」となります。dbtの仕様上テーブル化しないといけないので(依存関係の制御でも利用)、Procedure 実行後に処理完了を示すタイムスタンプを返す運用をとっています。Procedureがどこまで完了してるか調べやすくしています。以下のような形をとっています。
from databricks import sql # Import databricks-sql-connector to use SQL Warehouse resources instead of a cluster. from utils import calendar import pandas as pd .... def model(dbt, session): # tag definition dbt.config( tags=['<procedure_ parent_name>', 'procedure_name', 'batch', '<procedure_schema>'] ) # model dependencies refs = [dbt.ref("procedure_a")] # Control execution eligibility based on specific calendar conditions. if calendar.should_execute_today(dbt.config.get("calendar_table"), "bizday", session, dbt.config.get("calendar_date")): print('sp execute') try: # Procedure execution procedure_b(spark=session, params=params) # Return the procedure completion timestamp. return session.sql("SELECT current_timestamp() as createdtime, 'success' as status") except Exception as e: error_msg = f"Model procedure_b execution failed: {str(e)}" print(f"ERROR: {error_msg}") raise e # Return the procedure completion timestamp. return session.sql("SELECT current_timestamp() as createdtime, 'skip' as status")
データ比較評価
「コンバージョン評価プロセス全体の流れ」で紹介した通り、Procedureの実行結果をOracleの正解データと比較して評価します。また、Procedureの評価完了後はProcedureのフィードバックを反映した「評価環境」を正解データとし、インプットテーブルの評価も必要です。
Procedure単位での単体テストを行うため、事前にメタ情報としてProcedure単位で、書き込み処理をしてるテーブル一覧を整理しています。Procedure の実行によって更新されるテーブルを対象にデータ比較評価することで、Procedure 単体でのテストが可能になります。
write_tables: - table_name: table_a procedure: procedure_a schema: schema_a - table_name: table_b procedure: procedure_a schema: schema_a - table_name: table_a procedure: procedure_b schema: schema_b
Databricks Procedure の実行結果を評価するために、dbt Audit Helper を活用しています。カラムレベルおよびレコードレベルでのデータ比較が可能で、一致率も算出できるため、これらを指標として差分検知を行っています。dbtのエコシステムを使えるので、自前で作る必要もなく便利でした。
以下は行レベルの比較処理を行うサンプルとなります。「summarize」オプションを「TRUE」にするとデータ比較の一致率を確認することができます。「FALSE」にすると具体的にどの行で差分が発生しているかわかります。カラムレベルの比較等も可能です。データ比較結果はdbtモデルとしてテーブルに書き込まれます。
github.com
「dbt tag」は Procedure の実行単位に合わせて設計しています。これにより、Procedure 実行後に更新されたテーブルに対してデータ比較テストを実行できるようになっています。これらのファイルについても、先ほど紹介したメタ情報を利用して自動生成しています。
{{ config(
materialized='table',
tags=['<procedure_schema>', '<procedure_name>_test', '<procedure_parent_name>_test', '<table_name>_compare_queries_summarize', 'test_compare_queries_summarize']
) }}
{% set old_query %}
select
col_1,
col_2
from {{ var('catalog_table_compare_old') }}.<schema_name>.<table_name>
{% endset %}
{% set new_query %}
select
col_1,
col_2
from {{ var('catalog_table_compare_new') }}.<schema_name>.<table_name>
{% endset %}
{{ audit_helper.compare_queries(
a_query = old_query,
b_query = new_query,
summarize = true
) }}
以下のようにして、「dbt tag」で単体テストをしたいProcedureのデータ比較が可能です。データの比較結果はテーブルとして書き込まれます。
pipenv run dbt run --target=prod --select tag:<procedure_name>_test --threads 5
dbt Audit Helperのデータ比較結果に対して、dbt test を実行し、データ比較結果(一致率)に差分があった場合は検知できるようにしています。検知された結果は「人間」が確認し、LLM コンバージョンツールや評価環境構築処理、メタ情報などにフィードバックすることで、データレベルでの改善サイクルを回しています。dbt testファイルは以下のようにしています。
{{ config(
tags=['batch', '<schema>', 'test_<table_name>', '<procedure_name>_test', '<procedure_parent_name>_test']
) }}
select
1
from
{{ var('catalog_compare_new') }}.{{ var('schema_dbt_model') }}.<table_name>_<schema_name>_compare_queries_summarize
where
not (in_a = 'true' and in_b = 'true' and percent_of_total = 100)
以下のコマンドでテストできます。単体テスト、結合テスト等テストフェーズに合わせた「dbt tag」を指定し、柔軟にテストすることができます。
pipenv run dbt test --target=prod --select tag:<procedure_name>_test --threads 5
データ比較評価結果のフィードバック
「コンバージョン評価プロセス全体の流れ」で紹介した通り、データ比較評価結果のフィードバックはLLMのコンバージョンツールだけではなく、データの前処理、メタ情報、Oracle互換のUDF等多岐に渡ります。差分を確認した場合、どこに、どのように反映していくのが適切か人間が判断していく必要があります。運用としては、評価者が データ比較後に原因を調べ、Databricks Procedure を修正した Pull Request を作成し、レビュー担当者がその差分を確認したうえで、適切なフィードバック先を評価者と議論し、判断する運用としました。フィードバックに必要なコンテキスト情報は、基本的にはPull Request から取得する運用としています。
大変だったこと
プロンプトチューニングに伴うLLMのデグレ対応
データ評価のフィードバックをプロンプトへ反映していく過程で、コンテキストが肥大化し、精度がデグレする現象を確認しました。デグレが発生した場合は、GitHub 上でバージョンを戻すなどの対応を行いました。
また、以下のような運用と対策を取りました。
- ルール処理で対応可能なものは、極力ルールベースで処理する（コンテキストの肥大化を防ぐ）
- 変換結果は Pull Request でレビューする運用とする（GitHubの差分で明らかなデグレが発生していないか確認する）
- 評価済みの Procedure は再コンバート対象から除外する
- 変換処理を「LLM 処理」か「ルール処理」のいずれかに明示する（曖昧だとLLMの知識に頼ることになり、変換結果が不安定になる）
- 行数が多く、ビジネスロジックを含まないバッチ系 Procedure から優先的にコンバージョンを進める（ビジネスロジックは行数が少ないものの複雑な処理が増え、プロンプトのコンテキストが増えるため）
ルールベース処理と異なり、LLM は非決定的であるため不安定な側面があり、プロンプトのチューニングは手探りで進める必要がありました。そのため、プロンプト調整やデグレ対策には特に難しさを感じました。
データ比較評価
Syntaxレベルのテストは全Procedure1週間程度でクリアできましたが、データ比較はかなり大変でした。本記事でも一部ご紹介しましたが、データ比較で差分が出た場合、原因は多岐に渡ります。また、Procedureをテストするためのデータ準備等も自動化しているとはいえ、待ち時間はかかります。特に、複数の Procedure が同一のテーブルを更新している場合、原因調査はさらに難しくなります。正解データとして参照できるのは Oracle のバッチ完了後の断面のみであり、複数の Procedure によって更新された「最終更新結果」としかデータ比較ができないためです。その結果、どの Procedure の処理で問題が発生したのかを特定するのは困難になります。また、Procedureが更新するテーブルを評価することで、Procedureと関連のあるテーブル定義、UDF、前処理済みデータ、メタ情報、Procedure自体、どこに問題があっても検知はできますが、原因調査は大変です。できるだけシンプルにできるよう、Procedrue全体ではなく、単体で評価できる仕組みを作ったり、依存関係は既存のOracleの依存関係をそのまま利用するようにしました。
パフォーマンス改善
ご紹介したとおり、LLM コンバージョンツールを用いて Oracle の PL/SQL を Databricks Procedure にコンバージョンしました。しかし、OLTP を前提として作られている Oracle の PL/SQL を OLAP の Databricks に移行するにあたり、パフォーマンス面が大きな課題となりました。詳細については、以下の記事内の「Databricks Procedure導入課題」をご確認ください。
case-k.hatenablog.com
学び
コンバージョンプロセス全体を管理し、改善サイクルを高速に回す重要性
コンバージョンプロセス全体を管理し、評価結果を適切なプロセスへフィードバックしながら、継続的に改善していく仕組みの重要性を改めて感じました。コンバージョン評価結果を「人間」が確認後、フィードバック先は多岐に渡るので、全体のプロセスを管理し、改善サイクルを高速に回すことが大切です。全体ではなく、一部のプロセスにしかフィードバックできないと厳しかったです。LLM と dbt の各種機能を活用することで、PL/SQL のコンバージョンからデータ評価までを自動化でき、改善サイクルを高速に回せるようになりました。また、Databricks Procedure がサポートされたことで、SQL ウェアハウスから直接実行できるようになった点も、改善サイクルの高速化に寄与しています。さらに、PL/SQL のコンバージョンからデータ比較評価までをローカル環境で実行、検証できる仕組みを整えたことも、改善サイクルを迅速に回すうえで大きく寄与したと感じています。
LLMの力とソフトウェアエンジニアリングの重要性
LLM の登場により、これまでは諦めていたようなマイグレーションも、現実的な工数でできる時代になったと感じます（大変なことには変わりありませんが）。フィードバックを反映していくことで、マニュアル修正なしでデータ比較テストをパスするものが増え、コンバージョンペースは加速的に上がっていきました。ただし、LLM はそのまま使うだけでは十分な精度を発揮できず、現時点ではコンテキストエンジニアリングといった、ソフトウェアエンジニアリングが不可欠であることも実感しました。LLM をマイグレーションプロセスに組み込み、工数削減につながるレベルの成果を出すためには、十分に高い精度が求められます(特にデータ比較において差分が発生すると、その原因調査が必要となり、全体の進行ペースが大きく低下しました)。全体設計からコンテキストエンジニアリング、評価基盤、フィードバック改善プロセスの運用設計などソフトウェアエンジニアとしての能力が重要に思いました。
OSSへのコントリビュート機会
今回のマイグレーションを通じて、OSS にコントリビューションする機会がありました。dbt Python モデルの検証を進める中で、Job のアクセスコントロールに問題があることが分かりました。また、Notebook と Job の権限を分けて制御できなかったので、両者を分離できるよう機能追加および修正を行いました。
github.com
また、Databricks のコンソールから実行中のジョブをキャンセルした場合でも、成功として扱われてしまう問題もありました。これに対しても修正のための Pull Request を作成し、マージしてもらいました。
github.com
マージされるまではdbt Pythonモデル内でAPIを使い、パッチ対応しました。Databricksは他のOSSも含めてレビュー速度が非常に速いと感じます。組織としてOSSに注力されているのを感じます。また、機会等あればコントリビュートしていきたいです。
エージェントを活用したデータ比較評価の更なる自動化
ご紹介したとおり、Procedure の単体テストは、「人間」がローカル環境から各 dbt コマンドを実行して行ってきました。評価担当者は、データ比較で差分が検知された場合に原因を調査し、修正内容を反映した Pull Request を作成します。レビュー担当者がその Pull Request の内容を確認し、前述のとおり適切なコンバージョンプロセスへフィードバックします。
しかし、Claude Codeなどをデータ比較評価エージェントとして活用すれば、データ比較評価でもさらに自動化できた可能性があります。dbt評価基盤や各APIを「ツール」として扱い、dbt コマンドの実行から、データ比較評価、テーブルやファイルアクセスの伴う差分の原因調査、原因調査後の修正と再評価、修正結果を反映したPull Request の作成まで実現できるように思います。各コマンドはローカル環境から実行可能であり、差分発生時の原因調査方法も「人間」がエージェントのプロンプトにフィードバックできます。今回は制約もありそこまで踏み込めなかったのですが、PL/SQLのコンバージョンだけではなく、データ比較評価でも、エージェントを活用することで、データ比較評価を更に自動化できたように思います。
まとめ
LLMコンバージョンツール(LLM処理・ルール処理)と dbt 評価基盤の紹介をしました。LLMが使えなかったらコンバージョンは難しかったです。ただし、LLM はそのまま使うだけでは十分な精度を発揮できず、履歴管理やルール処理を含めた コンテキストエンジニアリング といった、ソフトウェアエンジニアリングが不可欠であることも実感しました。また、コンバージョン結果を評価するために評価基盤の整備が必要です。dbt を活用することで、依存関係の制御、タグを用いた単体テストから結合テストまで実行単位の切り替え、特定の日付断面を用いたローカル環境からの実行、テーブル定義や評価用データの準備、dbt Audit Helperを用いたデータ比較も同一のインターフェース上で実現できました。最後に、今回のコンバージョンを通じて、コンバージョンプロセス全体を管理し、評価結果を適切なプロセスへフィードバックしながら、継続的に改善していく仕組みが非常に大切だと思いました。LLM時代に評価基盤はセットですね。最後まで読んで頂きありがとうございました。良いお年を。
Oracle PL/SQL を Databricks にマイグレーションするための Databricks Procedure 導入Tips
Databricks Advent Calendar 2025 17日目の記事です。Oracle PL/SQLをDatabricksにマイグレーションする機会があり、Databricks Procedureが非常に便利だったので、導入Tipsをご紹介できればと思います。Databricks Procedureの活用を検討されてる方の参考になれば幸いです。
- Databricks Procedureとは
- Databricks Procedure が解決するマイグレーションの課題
- Databricks Procedure がサポートされて助かったところ
- Databricks Procedure導入課題
- Databricks Procedure関連で改善されると助かるところ
- まとめ
Databricks Procedureとは
Databricks上で実行できるProcedureで、SQL Scriptingをパッケージングし、Unity Catalogで管理することができます。ループ処理や IF 文による条件分岐など、PL/SQLで行っていたような処理をそのまま表現でき、マルチステートメントのトランザクション制御などにも対応しています。SQLで管理できる点も嬉しく、SQLエディタからSQLウェアハウスを利用した開発もできます。
CREATE PROCEDURE | Databricks on AWS
SQL scripting | Databricks on AWS
Databricks Procedure が解決するマイグレーションの課題
大規模かつ現役で変更が入り続けているOracle PL/SQLをリファクタし、SELECT 文に書き換えて移行するのはかなり厳しいです。
Databricks の Databricks Oracle Migration Guideにも現在はDatabricks Procedureへの移行も推奨していますが、従来はコンバージョンツールを使い、PySpark に変換する方法が推奨されていました。しかし、PySpark になってしまうとNotebook が必要になり、SQL で管理できなくなります。これまで Oracle をメンテナンスしてきたメンバーだと、Pythonの知見がなく、保守運用することも難しくなります。また、マルチステートメントのトランザクション制御も DB 側ではなく、アプリケーション側で担保する必要がありました。
Databricks で Procedure が利用可能になったことで、SQL管理できるようになり、運用担当者への引き継ぎもしやすく、SQLエディタからSQLウェアハウスを使った開発も可能です。マルチステートメントのトランザクション制御を含む、PL/SQLのマイグレーションに必要な様々な機能がサポートされています。
Databricks Procedure がサポートされて助かったところ
自動コンバージョン：構文エラーの検知と改善サイクルの高速化
構文エラーがある場合、「CREATE PROCEDURE 」を実行するとすぐに 「Syntax Error」 として検知されます。そのため、PL/SQL を自動コンバージョンする際に、変換結果を SQLウェアハウスにAPIで流すことで、構文エラーの評価と改善のサイクルを高速に回すことができました。PySpark では、spark.sql 内で SQL を文字列として記述することが多く、変数等も入ってくると、実際にジョブとして流してみないと、構文エラーの検知が難しかったように思います。また、「Syntax Error」後のデータ比較でも直接SQLエディタからも検証できたので、改善サイクルを高速化できました。
自動コンバージョンの詳細については、アドベントカレンダー25日目の記事 「Oracle PL/SQL→Databricks マイグレーションで作った LLMコンバージョンツールと dbt 評価基盤の紹介」 で紹介しています。ご興味があればぜひご覧ください。
case-k.hatenablog.com
冪等化：Multi-statement transactions
「ATOMIC」オプションを付与することで、マルチステートメントのトランザクション制御が可能です。この機能が使えない場合、アプリケーション側での制御が必要になりますが、DB側にトランザクション制御を任せることができるようになりました。バッチ処理でステートメントの途中で処理が落ちてしまっても、安全にリトライできます。
CREATE PROCEDURE procedure_name() LANGUAGE SQL SQL SECURITY INVOKER AS BEGIN ATOMIC END
まだ制限もあるため、トレードオフは考慮した方が良さそうです。例えば、「ATOMIC」オプションと例外処理「EXIT HANDLER」と併用することはできません。例外処理を使う場合、ラッパーとなる親のProcedureを作り、そこで例外処理と「ATOMIC」オプションを利用している、子Procedureの実行が必要になります。
UNSUPPORTED_FEATURE error condition | Databricks on AWS
※ 執筆時点ではまだPrivate Preview機能
同時実行制御：Session-local-temporary-tables
直接的なProcedureの機能ではないですが、Procedure内でSession-local-temporary-tablesを使うことで、セッション単位での一時的なテーブルを作成できます。一部、レポーティングツールを使い、複数のユーザーが同時実行しているProcedureがあります。PL/SQLでは同時実行制御に「global temporary table」を使い、対応していました。Session-local-temporary-tablesを使うことで、同時実行されても、競合やデータ不整合なく、一時テーブルへの書き込み処理が可能です。既存のPL/SQLの処理を大きく変えずに対応できるため助かりました。
CREATE temporary table temp_customers ( id INT, name STRING )
※ 執筆時点ではまだPrivate Preview機能
親子Procedureの分割実行
Procedure内で他のProcedureもCALL可能です。2万行を超えるような大規模な Procedure は、そのままでは大きすぎて各種リミットにも引っかかり、分割して対応する必要がありました。また、もともと親 Procedure から子 Procedure をCALLする構成になっていた処理も多く、その点は構造を保ったまま移行できます。(そのまま移行するかはプロジェクトの方針によります)先ほども触れましたが、「ATOMIC」オプションと例外処理「EXIT HANDLER」は併用できないため、Procedure を分割するなどの対応が必要になります。PySparkだった場合別途、新しいジョブを実行するなど必要で、Procedureのようにパッケージ化して呼び出せないと、難しかった部分なのではないかと思います。
CREATE PROCEDURE child_procedure() LANGUAGE SQL SQL SECURITY INVOKER AS BEGIN ATOMIC END CREATE PROCEDURE parent_procedure() LANGUAGE SQL SQL SECURITY INVOKER AS BEGIN CALL child_procedure() END
PL/SQL互換のパラメータ
PL/SQL同様、Databricks Procedureでも入力専用のパラメータ(IN)、入力・出力の両方に使用できるパラメータ(INOUT)、出力専用のパラメータ(OUT)が使えます。PL/SQLでも使われているため、Procedureの呼び出しもとを大きく変えずに移行できたので助かりました。
-- Demonstrate INOUT and OUT parameter usage. > CREATE OR REPLACE PROCEDURE add(x INT, y INT, OUT sum INT, INOUT total INT) LANGUAGE SQL SQL SECURITY INVOKER COMMENT 'Add two numbers' AS BEGIN SET sum = x + y; SET total = total + sum; END; > DECLARE sum INT; > DECLARE total INT DEFAULT 0; > CALL add(1, 2, sum, total); > SELECT sum, total; 3 3 > CALL add(3, 4, sum, total); 7 10
引用：https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-procedure#examples
Databricks Procedureでの SELECT 結果取得
OracleのPL/SQLだと、Procedure内のSELECT文実行結果を呼び出し元に返す場合、OUTパラメータとしてカーソルを返す必要があります。Databricksの場合はOUTパラメータの指定なしで、Procedureの最後に実行しているSELECT結果を返すようです。SELECT文以外も使えるテーブル関数のように利用できるので便利です。レポーティングツールなどで、ProcedureのSELECT結果を利用していたので、助かりました。Procedureがサポートされていなかったら、全てSELECT文のテーブル関数として、リファクタする必要があったと思います。
CREATE OR REPLACE PROCEDURE greeting(IN mode STRING COMMENT 'informal or formal') LANGUAGE SQL SQL SECURITY INVOKER AS BEGIN DECLARE msg STRING; SET msg = CASE WHEN mode = 'informal' THEN 'Hi!' WHEN mode = 'formal' THEN 'Pleased to meet you.' ELSE 'Hello!' END; INSERT INTO greeting_log VALUES (mode, msg); MERGE INTO greeting_summary t USING ( SELECT mode AS mode, COUNT(*) AS cnt FROM greeting_log WHERE mode = mode GROUP BY mode ) s ON t.mode = s.mode WHEN MATCHED THEN UPDATE SET cnt = s.cnt WHEN NOT MATCHED THEN INSERT (mode, cnt) VALUES (s.mode, s.cnt); SELECT mode AS input_mode, msg AS greeting_message; END; > CALL greeting('informal'); informal,Hi!
https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-procedure#examples
Databricks Procedure導入課題
新しい機能だったので導入にあたり不安はあったのですが、機能要件は全て満たすことができました。一方で、パフォーマンス面は問題となり、リファクタリングが必要なところもありました。
パフォーマンスの劣化
Databricks Procedureのまま移行したことで、パフォーマンスは問題となりました。
ループ処理対応
SQL Scripting でもループ処理はサポートされていますが、Oracle とは異なりインデックスが効かないため、Oracle では問題にならなかった「 N+1 問題」系のクエリは軒並みパフォーマンス問題に直面しました。 その結果、処理時間の大幅な悪化により処理が終わらず、ほとんどはループを使わず、JOIN等でリファクタしました。
全体的なパフォーマンス劣化
Oracle のテーブルは全体的にデータ量が少なく、インデックスも有効に機能していたため、高速に処理できていました。一方で、PL/SQLをDatabricks Procedure に移行した場合、ループだけでなく、読み取りや書き込み系の処理含め、全体的にパフォーマンスは劣化しました。Oracleのテーブルは100MBにも満たないスモールファイルも多く、リキッドクラスタリング等ファイルスキップでの高速化はできませんでした。バッチ系の処理についてはループ処理以外は大きな問題にはなりませんでしたが、レポーティングツールなどから同時実行され、リアルタイム性が求められるProcedure ではパフォーマンス劣化に伴う調整やOracle と同等のパフォーマンスを出すには事前計算し、モデリングが必要となりました。
Databricks Procedure関連で改善されると助かるところ
パフォーマンス問題
Databricksの設計的にに難しいことだとは思うのですが、Databricks Procedureマイグレーションした場合、同じ悩みにぶつかるように思います。ループ処理等許容できないレベル(1~10h)で遅くなってしまうため、複雑な処理をリファクタなしでマイグレーションできると、助かる方は非常に多くいるように思いました。
ODBCドライバのDatabricks Procedureサポート
VBAで作られているレポーティングツールからProcedureを実行する際にODBCドライバの利用を検討しました。検証したところ、ODBCドライバでProcedureを実行することはできましたが、Procedure内で実行したSELECT結果を取得することはできませんでした。
CREATE PROCEDURE | Databricks on AWS
ODBCドライバで取得するには一度テーブルとして書き出す必要があります。書き込んだテーブルを別途取得するという 多段のオペレーションが必要になり、すでにパフォーマンス課題を抱えている状況で、さらに処理が遅くなってしまうため断念しました。代わりの方法としてStatement Execution APIを利用することにしました。しかし、Statement Execution APIの場合、セッションが維持されないため、ODBCドライバで使えたSETなどは使えませんでした。また、Statement Execution APIはSQLウェアハウスに対してのみ使えるため、汎用クラスターなどは使えません。SQLウェアハウスのSpark Configを利用する場合、ワークスペースレベルでの変更が必要になるため、影響範囲を考えると難しいです。そのため、SETで対応予定だったタイムゾーンの制御を断念し、Procedure側で吸収することにしました。ODBCドライバのサポート、もしくはSQLウェアハウス単位でSpark Config設定ができると便利だと思いました。
UDFの上限緩和 or Oracle互換のUDF提供
直接Procedureとは関係ないのですが、マイグレーションを進める上で、Oracle互換のUDFをいくつか作りました。ProcdureからUDFの上限5つを超えてしまうことがあり、Spark Config側に以下の設定を加えました。基本的にSQLウェアハウスのリソースを使っていましたが、SQLウェアハウスは先ほども触れた通り、ワークスペースレベルでの適用になってしまうため、汎用クラスターで対応しました。UDFの上限緩和、もしくはSQLウェアハウス単位で、Spark Config設定ができると便利だと思いました。理想的にはOracle互換のUDFが提供されてると嬉しいです。
spark.databricks.safespark.externalUDF.plan.limit 30 spark.databricks.safespark.sandbox.trackMemory.enabled false spark.databricks.safespark.sandbox.size.default.mib 500
[UDF_MAX_COUNT_EXCEEDED] Exceeded query-wide UDF limit of 5 UDFs
Solved: [UDF_MAX_COUNT_EXCEEDED] Exceeded query-wide UDF l... - Databricks Community - 38120
SQLウェアハウス単位でのSpark Config設定
SQL ウェアハウス単位で Spark Config を設定できるようになると、Statement Execution APIでSETが使えない問題やUDFの上限緩和問題も解決するので、ぜひサポート頂きたい機能だと思いました。
SQL warehouse admin settings | Databricks on AWS
Atomicオプションと例外処理の併用
PL/SQL内では全て例外処理「EXIT HANDLER」を定義しており、エラーメッセージ等をテーブルに書き込んでいます。「Atomic」オプションを付与してしまうと、例外処理を使うことができなくなるため、Procedureの親子に分割して実行する必要がありました。「Atomic」オプションの導入メリットが大きかったので、分割しましたが、出来れば分割なしで実行できるようサポート頂けると嬉しいです。
まとめ
DatabricksでProcedureがサポートされたことで、マイグレーションが非常に楽になりました。一方で、DatabricksでProcedureへ移行する際にループ処理を含むパフォーマンスの問題はみなさんぶつかるところだと思うので、何か良いソリューションが出てくると嬉しいです。最後まで読んでいただきありがとうございました。
API GatewayとVPCエンドポイントを活用した、プライベートネットワークからパブリックネットワークへの接続手法
本稿では、API Gatewayをプロキシサーバとして活用し、プライベートネットワークからパブリックネットワークへ接続する方法をご紹介します。一般的にはNAT Gatewayを利用する構成が多いですが、セキュリティ要件などによりインターネットへのアクセスを厳格に管理する必要がある場合、API Gatewayの活用が有効です。また、最近ではLLMが注目される中、API Gatewayの利用クオータが引き上げられました。今後このような活用事例も増えてくるかもしれません。
aws.amazon.com
また、先日投稿した記事もLLMのサービングに関連する記事です。
www.case-k.jp
本記事では詳細な説明は省略いたしますが、Terraformによる定義例を共有いたします。同様の活用事例を検討される方の参考になれば幸いです。
Terraform
ネットワークリソース
まず、ネットワークリソースについて簡単に説明いたします。今回扱うVPCのリソースマップは以下の通りです。
図から、ネットワーク接続にインターネットゲートウェイやNAT Gatewayが存在しないことが確認できます。
これは、API Gatewayをプロキシサーバとしてインターナルアクセスする場合、これらのリソースを自前で用意する必要がないためです。
なお、参考までに、NAT Gatewayを用いた構成例は以下のようになります。
まずはネットワークリソースの作成から始めます。VPC環境からAPI Gatewayへのインターナルアクセスを実現するため、VPCエンドポイントを活用します。
resource "aws_vpc" "vpc" { cidr_block = "10.0.0.0/16" enable_dns_support = true enable_dns_hostnames = true tags = { "Name" = "vpc-for-api-gateway" } tags_all = { "Name" = "vpc-for-api-gateway" } } # resource "aws_subnet" "aws_subnet_private" { resource "aws_subnet" "subnet_private" { vpc_id = aws_vpc.vpc.id cidr_block = "10.0.1.0/24" tags = { "Name" = "private-subnet-for-api-gateway" } tags_all = { "Name" = "private-subnet-for-api-gateway" } }
ingressにはself = trueを設定し、同じセキュリティグループ内のリソース同士で通信できるようになります。NAT Gateway には不要ですが、API Gateway の VPC エンドポイントアクセスなど、セキュリティグループ内のリソース間で HTTPS 通信が必要な場合に有用です。アウトバウンドのトラフィックは全て許可します。
resource "aws_security_group" "security_group" { vpc_id = aws_vpc.vpc.id # ingress does not need for nat gateway but need for API gateway VPC endpoint access ingress { from_port = 443 to_port = 443 protocol = "tcp" self = true } egress { from_port = 0 to_port = 0 protocol = "-1" cidr_blocks = ["0.0.0.0/0"] } tags = { "Name" = "sg-private-subnet-for-api-gateway" } tags_all = { "Name" = "sg-private-subnet-for-api-gateway" } }
以下のコードは、Terraform を利用してプライベート API Gateway 用のルートテーブルおよび VPC エンドポイントを構築する例です。インターフェース型エンドポイントは、AWS PrivateLink を利用してプライベートにサービスに接続するためのものです。API Gatewayへのインターナル接続で利用します。
resource "aws_route_table" "private-route" { propagating_vgws = [] tags = { Name = "private-route-for-api-gateway" } tags_all = { Name = "private-route-for-api-gateway" } vpc_id = aws_vpc.vpc.id } resource "aws_route_table_association" "route_table_association" { subnet_id = aws_subnet.subnet_private.id route_table_id = aws_route_table.private-route.id } # ref # https://docs.aws.amazon.com/ja_jp/apigateway/latest/developerguide/apigateway-private-api-create.html resource "aws_vpc_endpoint" "vpc_endpoint" { vpc_id = aws_vpc.vpc.id service_name = "com.amazonaws.ap-northeast-1.execute-api" vpc_endpoint_type = "Interface" subnet_ids = [ aws_subnet.subnet_private.id ] security_group_ids = [ aws_security_group.security_group.id, ] private_dns_enabled = true tags = { Name = "vpc-endpoint-for-api-gateway" } tags_all = { Name = "vpc-endpoint-for-api-gateway" } }
API Gateway
以下はAPI Gatewayの定義例です。詳細は省略いたしますが、API Gatewayをプロキシサーバとして活用する構成となっています。APIエンドポイントのタイプはプライベートに設定し、VPCエンドポイントから接続できるようにしています。デプロイ後、API Gatewayのエンドポイントにリクエストを送ると、指定したエンドポイントへアクセスが可能です。
REST API の基本情報（名前やAPIキーの受け取り方法など）を設定し、リソースポリシーで「aws:SourceVpc」が指定の VPC と一致する場合のみ呼び出しを許可する条件を付与しています。また、エンドポイントを PRIVATE に設定し、特定の VPC エンドポイント（aws_vpc_endpoint.vpc_endpoint.id）と連携させています。
resource "aws_api_gateway_rest_api" "api_gateway_rest_api" { api_key_source = "HEADER" binary_media_types = [] body = null description = null disable_execute_api_endpoint = false fail_on_warnings = null minimum_compression_size = null name = "api-private-gw" parameters = null # policy = null policy = jsonencode({ Version = "2012-10-17" Statement = [ { Effect = "Allow" Principal = "*" Action = "execute-api:Invoke" Resource = "execute-api:/*" # Resource = "arn:aws:execute-api:ap-northeast-1:132483466678:l9jk54cpy0/*" Condition = { StringEquals = { "aws:SourceVpc" = aws_vpc.vpc.id } } } ] }) put_rest_api_mode = "overwrite" tags = {} tags_all = {} endpoint_configuration { types = ["PRIVATE"] vpc_endpoint_ids = [aws_vpc_endpoint.vpc_endpoint.id] } }
API のルートリソース直下に、パスパラメータ "{proxy+}" を持つリソースを作成します。
resource "aws_api_gateway_resource" "api_gateway_resource" { depends_on = [aws_api_gateway_rest_api.api_gateway_rest_api] # parent_id = aws_api_gateway_resource.api_gateway_resource_parent.id parent_id = aws_api_gateway_rest_api.api_gateway_rest_api.root_resource_id path_part = "{proxy+}" rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id }
上記リソースに対して HTTP のすべてのメソッド（ANY）を許可し、APIキー認証を必須としています。リクエストパラメータとして URL のパス部分を必須に設定。
resource "aws_api_gateway_method" "api_gateway_method" { depends_on = [aws_api_gateway_resource.api_gateway_resource] api_key_required = true authorization = "NONE" authorization_scopes = [] authorizer_id = null http_method = "ANY" operation_name = null request_models = {} request_parameters = { "method.request.path.proxy" = true } request_validator_id = null resource_id = aws_api_gateway_resource.api_gateway_resource.id rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id }
メソッド呼び出し成功時（200）のレスポンスモデルを定義。
resource "aws_api_gateway_method_response" "api_gateway_method_response" { rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id resource_id = aws_api_gateway_resource.api_gateway_resource.id http_method = aws_api_gateway_method.api_gateway_method.http_method status_code = "200" response_models = { "application/json" = "Empty" } }
バックエンドからのレスポンスを、JSON テンプレート（空のスキーマ）に変換する設定を行っています。
resource "aws_api_gateway_integration_response" "api_gateway_integration_response" { rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id resource_id = aws_api_gateway_resource.api_gateway_resource.id http_method = aws_api_gateway_method.api_gateway_method.http_method status_code = aws_api_gateway_method_response.api_gateway_method_response.status_code response_templates = { "application/json" = jsonencode( { "$schema" = "http://json-schema.org/draft-04/schema#" title = "Empty Schema" type = "object" } ) } }
API の呼び出しに必要な API キー（ここでは "test-key"）を作成。
resource "aws_api_gateway_api_key" "api_gateway_api_key" { customer_id = null description = null enabled = true name = "test-key" tags = {} tags_all = {} value = null # sensitive }
1日あたりのリクエスト上限（クォータ）やスロットリング（バースト・レート制限）を設定し、対象の API ステージ（"test"）を紐付けています。
resource "aws_api_gateway_deployment" "api_gateway_deployment" { depends_on = [aws_api_gateway_method.api_gateway_method] description = null rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id stage_description = null stage_name = null triggers = null variables = null }
作成した API キーと使用量プランを関連付け、API キー利用者に対して制限を適用します。
resource "aws_api_gateway_stage" "api_gateway_stage" { cache_cluster_enabled = false cache_cluster_size = null client_certificate_id = null deployment_id = aws_api_gateway_deployment.api_gateway_deployment.id description = null documentation_version = null rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id stage_name = "test" tags = {} tags_all = {} variables = {} xray_tracing_enabled = false } resource "aws_api_gateway_usage_plan" "api_gateway_usage_plan" { description = null name = "test-gw-plan" product_code = null tags = {} tags_all = {} api_stages { api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id stage = aws_api_gateway_stage.api_gateway_stage.stage_name } quota_settings { limit = 20 offset = 0 period = "DAY" } throttle_settings { burst_limit = 5 rate_limit = 5 } } resource "aws_api_gateway_usage_plan_key" "api_gateway_usage_plan_key" { key_id = aws_api_gateway_api_key.api_gateway_api_key.id key_type = "API_KEY" usage_plan_id = aws_api_gateway_usage_plan.api_gateway_usage_plan.id }
定義したメソッド（ANY）に対して、HTTP プロキシ統合を設定し、受け取ったリクエストを https://httpbin.org/get に転送します。キャッシュのキーとしてパスパラメータを利用し、タイムアウトなどの挙動も指定しています。
resource "aws_api_gateway_integration" "api_gateway_integration" { cache_key_parameters = ["method.request.path.proxy"] cache_namespace = aws_api_gateway_resource.api_gateway_resource.id connection_id = null connection_type = "INTERNET" content_handling = null credentials = null http_method = "ANY" integration_http_method = "GET" passthrough_behavior = "WHEN_NO_MATCH" request_parameters = { "integration.request.path.proxy" = "method.request.path.proxy" } request_templates = {} resource_id = aws_api_gateway_resource.api_gateway_resource.id rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id timeout_milliseconds = 5000 type = "HTTP" uri = "https://httpbin.org/get" }
Lambda
疎通確認用にLambdaをVPCにデプロイします。
import requests import json def lambda_handler(event, context): # ref # url = "https://httpbin.org/get" url = "https://<api-gw-id>.execute-api.ap-northeast-1.amazonaws.com/test/{proxy+}" # APIキーを指定 api_key = "api-key" headers = { "x-api-key": api_key } response = requests.get(url, headers=headers) print(response.text) return { "statusCode": response.status_code, "body": response.text }
Terraform
data "archive_file" "lambda_zip" { type = "zip" source_dir = "../app/lambda/python/package" output_path = "../app/lambda/python/lambda_function.zip" } resource "aws_lambda_function" "lambda_function_for_api_gateway" { depends_on = [data.archive_file.lambda_zip] function_name = "lambda_function_for_api_gateway" role = aws_iam_role.lambda_sample_function.arn handler = "lambda_function.lambda_handler" runtime = "python3.11" memory_size = 128 timeout = 5 source_code_hash = filebase64sha256(data.archive_file.lambda_zip.output_path) filename = data.archive_file.lambda_zip.output_path vpc_config { subnet_ids = [aws_subnet.subnet_private.id] security_group_ids = [aws_security_group.security_group.id] } }
terraform-provider-awsにコントリビュートする方法：Issueの起票からコントリビュートまで
terraform-provider-awsにコントリビュートする機会があったので、その過程を備忘録として記録しておきたいと思います。コントリビュートした内容としてはEventBridgeへの強制削除オプションの追加です。EventBridgeを触っている時に見つけたもので、issueを起票してコントリビュートしました。terraform-provider-awsへのコントリビュートを考えている方の参考になれば幸いです。
github.com
Issueの見つけ方
仕様の違和感に気づく
TerraformでCI/CDを構築する際、デプロイしたEventBridgeを削除しようとしたところ、以下のエラーが発生しました。EventBridgeのターゲットが存在している場合、強制的に削除ができない仕様のようです。この制約は不便に感じたため、APIの仕様を確認しました。私の場合はこのような違和感が、コントリビュートのきっかけになることが多いように思います。
│ Error: deleting EventBridge Rule (): ValidationException: Rule can't be deleted since it has targets. │ status code: 400, request id: <resource_id>
また、すでに多くのissueが起票されているので、もし興味のあるissueがあればここから探せます。
AWS API Referenceを確認する
APIで実装されていない機能はTerraformでも利用できないため、まずはEventBridgeのAPI仕様を確認します。今回はEventBridgeの強制削除がサポートされていることを確認しました。
docs.aws.amazon.com
Terraformのドキュメントを確認する
Terraformのドキュメントを確認し、EventBridgeにおける強制削除オプションがサポートされているか確認しました。今回はEventBridgeの強制削除オプションが未実装であることを確認しました。
github.com
Issueを起票する
Terraformで強制削除オプションがサポートされていないことを確認したため、Issueを起票しました。OSSによっては、Issueを作成する際に「Would you like to implement a fix?（修正を実装したいですか？）」という確認項目が含まれる場合があります。この場合、修正を自分で行いたい場合は、その旨を記載して意思を示します。
github.com
逆にそのような項目がないOSSに対してはテンプレートに項目を追加するのはどうか聞いてみるといいと思います。
過去に別のOSSで提案して、追加してもらいました。
github.com
コントリビュートする方法
Contributing Guideを確認する
READMEに記載されているContributing Guideを確認します。また、過去にマージされたPull Requestも参照し、コミュニティでの一般的なお作法や手順を把握します。確認すると、Pull Requestにはテストの実行結果を添付する必要があることが分かります。
Databricksでモデルサービングを迅速にデプロイするNotebook運用
本記事は、 Databricks - Qiita Advent Calendar 2024 - Qiitaシリーズ 2 の 25 日目の記事です。
モデルサービングをデプロイする際に使用しているNotebookテンプレートの運用をご紹介します。
モデルサービングはTerraformでデプロイすることも可能です。しかし、モデルの運用（再学習）などを考慮すると、Terraformで管理しているモデルのバージョンと実際に利用されているモデルのバージョンが一致しない場合があります。そこで、TerraformではNotebook Jobのみを管理し、Notebook内でモデルサービングのデプロイを行う運用を採用しています。
また、モデルサービングをデプロイする際には、モデルのロジック部分以外を共通化することでテンプレートとして管理しています。このテンプレートを利用することで、モデルの作成者はロジック部分のみを編集するだけで簡単にモデルサービングをデプロイできます。
Notebookテンプレート
以下のようなテンプレートを作成しています。「EDITABLE 」となっている箇所がモデル作成者側で修正する箇所になります。利用者はテンプレートをコピーし、「EDITABLE 」となっているモデルのロジック部分を修正します。「EDITABLE 」の前後にある「pre_hook」と「post_hook」は共通処理となります。後ほど紹介しますが、Notebookに渡されたパラメータに基づいてモデルやモデルサービングの登録や更新、削除などの処理をしています。「EDITABLE」となっている箇所も別ファイルとして管理した方がテンプレートはシンプルになりますが、Databricks上でモデルの実行結果等可視化して確認しやすいようこのような運用を採用しています。
# Databricks notebook source # MAGIC %run ./model_serving # COMMAND ---------- import json import mlflow mlflow.set_registry_uri("databricks-uc") # COMMAND ---------- params_string = dbutils.widgets.get("params") params = json.loads(params_string) print(f'params: {params}') # COMMAND ---------- # MAGIC %python # MAGIC pre_hook(params) # COMMAND ---------- ################ EDITABLE ################ ################ START ################ from sklearn import datasets from sklearn.ensemble import RandomForestClassifier model = params['model'] artifact_path = model['artifact_path'] endpoint = params['endpoint'] model_name = endpoint['config']['served_entities'][0]['entity_name'] experiment_path = model['experiment_path'] mlflow.set_experiment(experiment_path) with mlflow.start_run(): # Train a sklearn model on the iris dataset X, y = datasets.load_iris(return_X_y=True, as_frame=True) clf = RandomForestClassifier(max_depth=7) clf.fit(X, y) # Take the first row of the training dataset as the model input example. input_example = X.iloc[[0]] # Log the model and register it as a new version in UC. mlflow.sklearn.log_model( sk_model=clf, artifact_path=artifact_path, # The signature is automatically inferred from the input example and its predicted output. input_example=input_example, registered_model_name=model_name, ) ################ END ################ # COMMAND ---------- # MAGIC %python # MAGIC post_hook(params)
次にテンプレートで参照している「pre_hook」と「post_hook」について紹介します。以下のコードを見るとわかりますが、Notebookのパラメータに基づいてモデルやモデルサービングの作成や更新、削除等を行っています。「pre_hook」と「post_hook」の処理から見ていただくと雰囲気が掴めると思います。
# Databricks notebook source from mlflow.deployments import get_deploy_client from mlflow.tracking import MlflowClient import time from distutils.util import strtobool # COMMAND ---------- def pre_hook(params): model = params['model'] force_delete = bool(strtobool(params['force_delete'])) endpoint = params['endpoint'] endpoint_name = endpoint['endpoint_name'] config = endpoint['config'] model_name = config['served_entities'][0]['entity_name'] model_version = config['served_entities'][0]['entity_version'] print(f'codfig: {config}') print(f'force_delete: {force_delete}') if model_registered_exists(model_name) and force_delete: delete_registered_model(model_name) print('skip delete_registered_model') if endpoint_exists(endpoint_name) and force_delete: delete_model_serving_endpoint(endpoint_name) dbutils.notebook.exit("Model serving endpoint deleted. Exiting the notebook.") print('skip delete_model_serving_endpoint') if model_registered_exists(model_name) and endpoint_exists(endpoint_name) and model_version != 'latest': update_model_serving_endpoint(endpoint_name, config) wait_for_endpoint_ready(endpoint_name) dbutils.notebook.exit(f'Updated model endpoint version: {model_version}. Exiting the notebook') print('skip update_model_serving_endpoint') def post_hook(params): endpoint = params['endpoint'] endpoint_name = endpoint['endpoint_name'] config = endpoint['config'] model_name = config['served_entities'][0]['entity_name'] model_version = config['served_entities'][0]['entity_version'] print(f'codfig: {config}') if model_version == 'latest': model_version = get_model_registered_latest_version(model_name) config['served_entities'][0]['entity_version'] = model_version print(f'model version: {model_version}') # Check if the endpoint exists if endpoint_exists(endpoint_name): print(f"Endpoint '{endpoint_name}' exists.") wait_for_endpoint_ready(endpoint_name) print(f'update_model_serving_endpoint') update_model_serving_endpoint(endpoint_name, config) else: print(f"Endpoint '{endpoint_name}' does not exist.") print(f'create_model_serving_endpoint') create_model_serving_endpoint(endpoint_name, config) wait_for_endpoint_ready(endpoint_name) print('end') # Model Registerd def delete_registered_model(model_name): try: client = MlflowClient() client.delete_registered_model(name=model_name) print(f"Model '{model_name}' has been deleted.") except Exception as e: print(f"Error deleting model '{model_name}': {e}") raise def model_registered_exists(model_name): try: client = MlflowClient() registered_models = client.search_registered_models() return any(model.name == model_name for model in registered_models) except Exception as e: print(f"Error checking if model '{model_name}' is registered in Unity Catalog: {e}") return False def get_model_registered_latest_version(model_name): try: client = MlflowClient() print(f'model_name:{model_name}') client = MlflowClient() model_version_infos = client.search_model_versions("name = '%s'" % model_name) print(f'model_version_infos:{model_version_infos}') latest_version = max([int(model_version_info.version) for model_version_info in model_version_infos]) return latest_version except Exception as e: print(f"Error get_model_registered_latest_version: {e}") raise # Model serving def delete_model_serving_endpoint(endpoint_name): try: deploy_client = get_deploy_client("databricks") deploy_client.delete_endpoint(endpoint=endpoint_name) print(f"Model serving endpoint '{endpoint_name}' has been deleted.") except Exception as e: print(f"Error deleting model serving endpoint '{endpoint_name}': {e}") raise def create_model_serving_endpoint(name, config): try: deploy_client = get_deploy_client("databricks") deploy_client.create_endpoint(name=name,config=config) print(f"Model serving endpoint '{name}' has been created. config: {config}") except Exception as e: print(f"Error createing model serving endpoint '{name}': {e}") raise def update_model_serving_endpoint(endpoint_name, config): try: deploy_client = get_deploy_client("databricks") deploy_client.update_endpoint(endpoint=endpoint_name,config=config) print(f"Model serving endpoint '{endpoint_name}' has been updated. config: {config}") except Exception as e: print(f"Error updating model serving endpoint '{endpoint_name}': {e}") raise def endpoint_exists(endpoint_name): try: deploy_client = get_deploy_client("databricks") endpoints = deploy_client.list_endpoints() return any(endpoint['name'] == endpoint_name for endpoint in endpoints) except Exception as e: print(f"Error checking if endpoint exists '{endpoint_name}': {e}") return False def get_endpoint_status(endpoint_name): try: deploy_client = get_deploy_client("databricks") endpoint_status = deploy_client.get_endpoint(endpoint=endpoint_name) return endpoint_status except Exception as e: print(f"Error getting status for endpoint '{endpoint_name}': {e}") raise def wait_for_endpoint_ready(endpoint_name, timeout=1000, interval=30): start_time = time.time() while time.time() - start_time < timeout: try: # https://docs.databricks.com/api/workspace/servingendpoints/get status = get_endpoint_status(endpoint_name) if status['state']['ready'] == 'READY': print(f"Endpoint '{endpoint_name}' is ready.") if status['state']['config_update'] == 'NOT_UPDATING': print(f"Endpoint '{endpoint_name}' is currently NOT_UPDATING.") return else: print(f"Endpoint '{endpoint_name}' is currently being updated. Waiting...") except Exception as e: if "RESOURCE_CONFLICT" in str(e): print(f"Endpoint '{endpoint_name}' is currently being updated. Waiting...") else: raise time.sleep(interval) raise TimeoutError(f"Endpoint '{endpoint_name}' is not ready after {timeout} seconds.")
Databricks Job(Terraform)
Notebook JobはTerraformを使用してデプロイしており、パラメータに関する情報は以下のYAMLファイルに定義されています。このYAMLの設定に基づき、TerraformでNotebook Jobを動的にデプロイしています。
jobs: wf_test_model_endpoint_tmpl: name: wf_test_model_endpoint_tmpl domain_tag: test notebook_path: "notebook/model_serving/model_endpoint.tmpl" job_params: { "force_delete": "false", "model": { "experiment_path": "/Shared/common_model_experiments/sample", "artifact_path": "model", }, "endpoint": { "endpoint_name": "workspace-model-endpoint", "config": { "served_entities": [ { "name": "iris_model_serving", "entity_name": "${Env}_data_science.<project>.sample_model", "entity_version": "latest", # latest or version number for rollback "workload_size": "Small", "scale_to_zero_enabled": "true", }, ], "auto_capture_config": { "catalog_name": "${Env}_catalog", "schema_name": "sample_schema, "enabled": "true" } }, }, } clusters: { spark_version: "15.3.x-cpu-ml-scala2.12", node_type_id: "i3.2xlarge", driver_node_type_id: "i3.2xlarge", autoscale: { min_workers: 2, max_workers: 10 }, } template_path: "../jobs/template/model_serving_template.json" git_url: "https://git-codecommit.ap-northeast-1.amazonaws.com/v1/repos/<repository_name>"
モデルサービング用に作成したNotebook Jobのテンプレートは以下のとおりです。このテンプレートでは、YAMLで定義されたパラメータを活用して動的にJobの定義を構築しています。
{ "name": "${name}", "email_notifications": { "no_alert_for_skipped_runs": false }, "notification_settings": { "no_alert_for_canceled_runs": true }, "webhook_notifications": {}, "timeout_seconds": 0, "max_concurrent_runs": 1, "tags": { "product": "${domain_tag}" }, "parameters": [ { "name": "params", "default": "${params}" } ], "job_clusters": [ { "job_cluster_key": "job_cluster_key_${env}", "new_cluster": { "spark_version": "${spark_version}", "node_type_id": "${node_type_id}", "driver_node_type_id": "${driver_node_type_id}", "policy_id": "${policy_id}", "autoscale": { "min_workers": "${min_workers}", "max_workers": "${max_workers}" }, "aws_attributes": { "first_on_demand": "${first_on_demand}" } } } ], "tasks": [ { "task_key": "deploy_model_serving", "max_retries": 0, "notebook_task": { "notebook_path": "${notebook_path}", "source": "GIT" }, "job_cluster_key": "job_cluster_key_${env}", "libraries": [ { "pypi": { "package": "mlflow-skinny[databricks]>=2.5.0" } } ] } ], "git_source": { "git_url": "${git_url}", "git_provider": "awscodecommit", "git_tag": "t_${env}" }, "format": "MULTI_TASK" }
動的に構築されたJobの情報は、以下のTerraformで定義されたJobに渡されています。この仕組みでは、モデルサービング以外のジョブリソースも含めて共通の定義を活用し、効率的な管理を実現しています。各ジョブの差分はlocalsで吸収し、local.job-association-mapを用いて全てのジョブリソース情報を一元的に受け取る構造となっています。
resource "databricks_job" "job" { for_each = local.job-association-map depends_on = [ databricks_cluster.shared_dbx_cluster ] name = each.value.name timeout_seconds = each.value.timeout_seconds max_concurrent_runs = each.value.max_concurrent_runs git_source { url = lookup(each.value, "git_source", null) != null ? each.value.git_source.git_url : null provider = lookup(each.value, "git_source", null) != null ? each.value.git_source.git_provider : null tag = lookup(each.value, "git_source", null) != null ? each.value.git_source.git_tag : null } tags = { product = lookup(each.value, "tags", null) != null ? each.value.tags.product : local.tags.product } email_notifications { no_alert_for_skipped_runs = lookup(each.value.email_notifications, "no_alert_for_skipped_runs", null) != null ? each.value.email_notifications.no_alert_for_skipped_runs : null on_start = lookup(each.value.email_notifications, "on_start", []) != [] ? each.value.email_notifications.on_start : [] on_success = lookup(each.value.email_notifications, "on_success", []) != [] ? each.value.email_notifications.on_success : [] on_failure = lookup(each.value.email_notifications, "on_failure", []) != [] ? each.value.email_notifications.on_failure : local.on_failure } format = each.value.format dynamic "trigger" { for_each = { for key, val in each.value : key => val if key == "trigger" && val != null } content { pause_status = trigger.value.pause_status file_arrival { url = trigger.value.file_arrival.url } } } # use existing cluster instead of new_cluster.This will be used for IDBCDB,To import existing resources. dynamic "job_cluster" { for_each = { for key, val in each.value.job_clusters : key => val if lookup(val, "new_cluster", null) != null } content { job_cluster_key = each.value.job_clusters[0].job_cluster_key new_cluster { spark_version = lookup(each.value.job_clusters[0].new_cluster, "spark_version", null) != null ? each.value.job_clusters[0].new_cluster.spark_version : local.clusters.spark_version node_type_id = lookup(each.value.job_clusters[0].new_cluster, "node_type_id", null) != null ? each.value.job_clusters[0].new_cluster.node_type_id : local.clusters.node_type_id driver_node_type_id = lookup(each.value.job_clusters[0].new_cluster, "driver_node_type_id", null) != null ? each.value.job_clusters[0].new_cluster.driver_node_type_id : local.clusters.driver_node_type_id policy_id = lookup(each.value.job_clusters[0].new_cluster, "policy_id", null) != null ? each.value.job_clusters[0].new_cluster.policy_id : local.clusters.policy_id runtime_engine = lookup(each.value.job_clusters[0].new_cluster, "runtime_engine", null) != null ? each.value.job_clusters[0].new_cluster.runtime_engine : local.clusters.runtime_engine spark_conf = lookup(each.value.job_clusters[0].new_cluster, "spark_conf", null) != null ? each.value.job_clusters[0].new_cluster.spark_conf : null autoscale { min_workers = lookup(each.value.job_clusters[0].new_cluster.autoscale, "min_workers", null) != null ? each.value.job_clusters[0].new_cluster.autoscale.min_workers : local.clusters.autoscale.min_workers max_workers = lookup(each.value.job_clusters[0].new_cluster.autoscale, "max_workers", null) != null ? each.value.job_clusters[0].new_cluster.autoscale.max_workers : local.clusters.autoscale.max_workers } aws_attributes { first_on_demand = lookup(each.value.job_clusters[0].new_cluster.aws_attributes, "first_on_demand", null) != null ? each.value.job_clusters[0].new_cluster.aws_attributes.first_on_demand : local.clusters.aws_attributes.first_on_demand availability = lookup(each.value.job_clusters[0].new_cluster.aws_attributes, "availability", null) != null ? each.value.job_clusters[0].new_cluster.aws_attributes.availability : local.clusters.aws_attributes.availability instance_profile_arn = lookup(each.value.job_clusters[0].new_cluster.aws_attributes, "instance_profile_arn", null) != null ? each.value.job_clusters[0].new_cluster.aws_attributes.instance_profile_arn : local.clusters.aws_attributes.instance_profile_arn zone_id = lookup(each.value.job_clusters[0].new_cluster.aws_attributes, "zone_id", null) != null ? each.value.job_clusters[0].new_cluster.aws_attributes.zone_id : local.clusters.aws_attributes.zone_id spot_bid_price_percent = lookup(each.value.job_clusters[0].new_cluster.aws_attributes, "spot_bid_price_percent", null) != null ? each.value.job_clusters[0].new_cluster.aws_attributes.spot_bid_price_percent : local.clusters.aws_attributes.spot_bid_price_percent } custom_tags = each.value.job_clusters[0].new_cluster.custom_tags } } } dynamic "notification_settings" { for_each = { for key, val in each.value : key => val if key == "notification_settings" && val != null } content { no_alert_for_skipped_runs = lookup(notification_settings.value, "no_alert_for_skipped_runs", null) != null ? notification_settings.value.no_alert_for_skipped_runs : null no_alert_for_canceled_runs = lookup(notification_settings.value, "no_alert_for_canceled_runs", null) != null ? notification_settings.value.no_alert_for_canceled_runs : null } } dynamic "schedule" { for_each = { for key, val in each.value : key => val if key == "schedule" && val != null } content { pause_status = schedule.value.pause_status quartz_cron_expression = schedule.value.quartz_cron_expression timezone_id = schedule.value.timezone_id } } dynamic "parameter" { for_each = contains(keys(each.value), "parameter") ? each.value["parameter"] : [] content { default = parameter.value.default name = parameter.value.name } } dynamic "queue" { for_each = { for key, val in each.value : key => val if key == "queue" && val != {} } content { # enabled = lookup(queue.value, "enabled", null) != null ? queue.value.enabled : false enabled = queue.value.enabled } } dynamic "task" { for_each = each.value.tasks content { task_key = task.value.task_key job_cluster_key = lookup(task.value, "job_cluster_key", null) != null ? task.value.job_cluster_key : null existing_cluster_id = lookup(task.value, "existing_cluster_id", null) != null ? task.value.existing_cluster_id : null max_retries = contains(keys(task.value), "max_retries") ? task.value["max_retries"] : local.max_retries min_retry_interval_millis = contains(keys(task.value), "min_retry_interval_millis") ? task.value["min_retry_interval_millis"] : local.min_retry_interval_millis run_if = lookup(task.value, "run_if", null) != null ? task.value.run_if : null dynamic "notebook_task" { for_each = { for key, val in task.value : key => val if key == "notebook_task" } content { notebook_path = notebook_task.value.notebook_path base_parameters = lookup(notebook_task.value, "base_parameters", null) != null ? notebook_task.value.base_parameters : {} source = notebook_task.value.source } } dynamic "depends_on" { for_each = contains(keys(task.value), "depends_on") ? task.value["depends_on"] : [] content { task_key = depends_on.value.task_key outcome = lookup(depends_on.value, "outcome", null) != null ? depends_on.value.outcome : null } } dynamic "dbt_task" { for_each = { for key, val in task.value : key => val if key == "dbt_task" } content { project_directory = task.value.dbt_task.project_directory commands = task.value.dbt_task.commands schema = task.value.dbt_task.schema warehouse_id = task.value.dbt_task.warehouse_id catalog = task.value.dbt_task.catalog } } dynamic "spark_python_task" { for_each = { for key, val in task.value : key => val if key == "spark_python_task" } content { parameters = task.value.spark_python_task.parameters python_file = task.value.spark_python_task.python_file source = task.value.spark_python_task.source } } dynamic "condition_task" { for_each = { for key, val in task.value : key => val if key == "condition_task" } content { left = task.value.condition_task.left op = task.value.condition_task.op right = task.value.condition_task.right } } dynamic "library" { for_each = contains(keys(task.value), "libraries") ? task.value["libraries"] : [] content { pypi { package = task.value.libraries[0].pypi.package } } } timeout_seconds = lookup(task.value, "timeout_seconds", null) != null ? task.value.timeout_seconds : null dynamic "email_notifications" { for_each = { for key, val in task.value : key => val if key == "email_notifications" && val != {} } content { on_success = lookup(email_notifications.value, "on_success", null) != null ? email_notifications.value.on_success : null on_start = lookup(email_notifications.value, "on_start", null) != null ? email_notifications.value.on_start : null on_failure = lookup(email_notifications.value, "on_failure", null) != null ? email_notifications.value.on_failure : null } } dynamic "notification_settings" { for_each = { for key, val in task.value : key => val if key == "notification_settings" && val != {} } content { alert_on_last_attempt = lookup(notification_settings.value, "alert_on_last_attempt", null) != null ? notification_settings.value.alert_on_last_attempt : false no_alert_for_canceled_runs = lookup(notification_settings.value, "no_alert_for_canceled_runs", null) != null ? notification_settings.value.no_alert_for_canceled_runs : false no_alert_for_skipped_runs = lookup(notification_settings.value, "no_alert_for_skipped_runs", null) != null ? notification_settings.value.no_alert_for_skipped_runs : false } } } } }
まとめ
モデルサービングのデプロイはモデルの再学習等を考慮して、Terraformで直接デプロイするのではなく、Notebook Jobを活用しています。Notebookはロジック部分を除き共通化できるので、テンプレート化して利用者に提供しています。利用者はテンプレートを活用することで、モデルを作成したのち高速にデプロイできる環境を利用できます。
Databricks × dbt 運用Tips：失敗したモデルだけを効率的にリトライする方法
本記事は、 Databricks - Qiita Advent Calendar 2024 - Qiitaシリーズ 3 の 24 日目の記事です。
データ基盤を運用する際、依存関係を考慮しながら、失敗したモデルやその依存関係のあるモデルのみを再実行したい場合があります。本記事では、Databricksとdbtを活用したリトライ方法についてご紹介します。
Case 1：失敗したモデルと依存関係のある後続モデルのリトライ
dbtの機能である「dbt retry」を活用し、失敗したモデルとその依存関係にある後続処理を効率的に再実行する方法をご紹介します。
dbtの実行結果は「run_results.json」で確認可能です。「dbt retry」では、この「run_results.json」に記録されたログを基に、失敗したモデルのみを再実行することができます。
dbt Cloudでは、失敗したモデルのみを再実行する機能が提供されています。一方、OSS版のdbt Coreを利用している場合、ログが欠損しないようストレージに永続化する仕組みが必要です。しかし、Databricksが提供する標準的なdbt処理ではこの要件に対応できないため、専用のdbt用共通モジュールを作成し、Notebook内でdbtを実行する運用を採用しています。
具体的には、以下のコードを使用して運用を行っています。Notebook Jobで渡されたパラメータに基づき、dbt runまたはdbt retryを制御します。また、永続化先のストレージで障害が発生した場合に実行ログを失う可能性を考慮し、ログは標準出力にも記録しています。
# Databricks notebook source # COMMAND ---------- import os import time from datetime import datetime import pytz import json env = dbutils.widgets.get("env") dbt_tag = dbutils.widgets.get("dbt_tag") threads = dbutils.widgets.get("threads") dbt_dbfs_state_dir = dbutils.widgets.get("dbt_dbfs_state_dir") dbt_tmp_dir = f'/tmp-{time.time()}' dbt_project_dir = f'{dbt_tmp_dir}/dbt/config/{env}' dbt_profile_dir = f'{dbt_tmp_dir}/dbt/config' dbt_state_dir = f'{dbt_project_dir}/target' os.environ['env'] = env os.environ['dbt_tag'] = dbt_tag os.environ['threads'] = threads os.environ['DBT_TEMP_DIR'] = dbt_tmp_dir os.environ['DBT_PROFILES_DIR'] = dbt_profile_dir os.environ['DBT_PROJECT_DIR'] = dbt_project_dir os.environ['DBT_STATE_DIR'] = dbt_state_dir os.environ['DBT_ENV_SECRET_TOKEN'] = dbutils.secrets.get(f'<secret>', f'<secret-token>') # COMMAND ---------- # MAGIC %sh # MAGIC set -eu # MAGIC ls -ltr ../ # MAGIC mkdir -p ${DBT_STATE_DIR} # MAGIC cp -r ../dbt ${DBT_TEMP_DIR} # COMMAND ---------- # This params are used to run dbt command with the custome date dbt_retry = dbutils.widgets.get('dbt_retry') dbt_command = f"dbt run --select {dbt_tag} --target={env} --threads {threads}" if dbt_retry != 'false': dbutils.fs.cp(f'{dbt_dbfs_state_dir}/run_results.json', f'file:{dbt_state_dir}/run_results.json') dbt_command = f"dbt retry --target={env} --threads {threads}" print(f'dbt command: {dbt_command}') os.environ['DBT_COMMAND'] = dbt_command # COMMAND ---------- # MAGIC %sh # MAGIC set -eu # MAGIC cd ${DBT_TEMP_DIR} # MAGIC dbt deps # MAGIC ${DBT_COMMAND} # COMMAND ---------- file_path = f'{dbt_state_dir}/run_results.json' try: with open(file_path, 'r') as file: data = json.load(file) print(json.dumps(data)) if dbt_retry != 'false': # Remove crc file to avoid error when copying to dbfs dbutils.fs.rm( f'file:{dbt_state_dir}/.run_results.json.crc') dbutils.fs.cp(f'file:{file_path}', f'{dbt_dbfs_state_dir}/run_results.json') for result in data['results']: if result['status'] == 'error': raise ValueError("Error detected in results") except Exception as e: raise e
Case 2：データ不整合が判明したモデルのリトライ
次に想定されるシナリオとして、モデルの実行結果に不整合が発生し、再度成功したモデルを実行したい場合があります。このようなユースケースでは、dbtのタグ機能を活用します。以下に、モデルmodel_aおよびmodel_bを実行するケースを示します。
dbt run --select tag:model_a,model_b
```
まとめ
dbtで失敗したモデルを再実行する際には、dbt retryを利用しています。ログの損失を防ぐために、ログを永続化する仕組みを整えた上で、Notebookを通じてdbtを実行しています。また、既に成功しているモデルをリトライしたい場合には、dbtのタグ機能を活用して対応しています。
DatabricksのOSS(terraform, dbt)にコントリビュートした話
この記事は、Databricksとdbtのアドベントカレンダー2023の13日めの記事です。
ここ半年ほどDatabricksを使い始めて、最近terraformやdbtにコントリビュートする機会があったので、その紹介をします。色々見つかって楽しいので来年はもっとコントリビュートしていきたいです。
| Open Source Repositories | Title & Url | Status |
| terraform-provider-databricks | Sort based on the Task Key specified in the 'Depends On' field https://github.com/databricks/terraform-provider-databricks/pull/3000 | merged |
| terraform-provider-databricks | Add Test Code for Sorting Tasks in the 'Depends On' Field of a Job https://github.com/databricks/terraform-provider-databricks/pull/3183 | merged |
| terraform-provider-databricks | Added an item to check if the ticket opener wants to do a bug fix https://github.com/databricks/terraform-provider-databricks/pull/3020 | merged |
| dbt-databricks | Fix dbt incremental_strategy behavior by fixing schema table existing check https://github.com/databricks/dbt-databricks/pull/530 | merged |
| dbt-databricks | Add DBT_DATABRICKS_UC_INITIAL_CATALOG Option https://github.com/databricks/dbt-databricks/pull/537 | merged |
| dbt-databricks | Eliminate the conversion of the schema to lowercase for schema-related test https://github.com/databricks/dbt-databricks/pull/541 | merged |
| dbt-databricks | Add schema option for testing https://github.com/databricks/dbt-databricks/pull/548 / https://github.com/databricks/dbt-databricks/pull/538 | merged |
Databricksはまだ、発展途上の部分もありOSSに対してコントリビュートできる機会が多く楽しめています。来年は開発や運用で見つけたバグだけではなく、issueも積極的に漁りながら貢献していきたいです。
Javaのバージョン管理 備忘録
jenv install
brew install jenv echo "if which jenv > /dev/null; then eval "$(jenv init -)"; fi" >> ~/.zshrc source ~/.zshrc
M1 Macの場合
sudo softwareupdate --install-rosetta # M1で入れるのに必要
JDK Install
brew tap AdoptOpenJDK/openjdk brew install --cask adoptopenjdk8 brew install --cask adoptopenjdk /usr/libexec/java_home -v 1.8 /Library/Java/JavaVirtualMachines/adoptopenjdk-8.jdk/Contents/Home /usr/libexec/java_home -v 16 /Library/Java/JavaVirtualMachines/adoptopenjdk-16.jdk/Contents/Home
JenvにJDKを追加
jenv add /Library/Java/JavaVirtualMachines/adoptopenjdk-8.jdk/Contents/Home
jenv add /Library/Java/JavaVirtualMachines/adoptopenjdk-16.jdk/Contents/Home
```
Jenvに追加されたバージョンを確認
jenv versions * system (set by /Users/keisuke.taniguchi/.jenv/version) 1.8 1.8.0.292 16.0 16.0.1 openjdk64-1.8.0.292 openjdk64-16.0.1
利用するバージョンを選択
jenv global 1.8 java -version openjdk version "1.8.0_292" OpenJDK Runtime Environment (AdoptOpenJDK)(build 1.8.0_292-b10) OpenJDK 64-Bit Server VM (AdoptOpenJDK)(build 25.292-b10, mixed mode)
ref
Macで多様なJavaバージョンとディストリビューションを管理：HomebrewとjEnvの活用 #homebrew - Qiita
Java 11
brew install --cask temurin11 jenv add /Library/Java/JavaVirtualMachines/temurin-11.jdk/Contents/Home
Java21
```
brew install --cask temurin21
jenv add /Library/Java/JavaVirtualMachines/temurin-21.jdk/Contents/Home
jenv global 21.0
keisuke.taniguchi@keisuketaniguchinoMacBook-Pro api % jenv versions
system
1.8
1.8.0.292
11.0
11.0.26
21.0 (set by /Users/keisuke.taniguchi/.jenv/version)
21.0.6
openjdk64-1.8.0.292
temurin64-11.0.26
temurin64-21.0.6
```
BigQueryに書き込まれたSQL Serverの変更ログを用いて、変更のあったPKの最新データと変更前のデータを取得する方法
ZOZO Advent Calendar 2022 カレンダー25日目の記事です
qiita.com
BigQueryに書き込まれたSQL Serverの変更追跡ログを使って、変更のあったPKの変更ログと変更前のログを取得する方法をご紹介します。
ZOZOではSQL Serverの変更追跡機能を使い、変更のあったPKの最新のレコードをBigQueryに連携しています。SQL Serverの変更追跡では以下のようなクエリを実行して、変更のあったPKとPKに紐づく最新のデータを取得できます。
SELECT a.SYS_CHANGE_OPERATION as changetrack_type, a.SYS_CHANGE_VERSION as changetrack_ver, #{columns} FROM CHANGETABLE(CHANGES #{@tablename}, @前回更新したバージョン) AS a LEFT OUTER JOIN #{@tablename} ON a.#{@primary_key} = b.#{@primary_key}
SQL ServerなどDBの変更を追跡する機能はCDC(Change Data Capture)と呼ばれており、DBで変更のあったデータを全て連携するログベースのCDCから、クエリで変更データをポーリング するCDCなどがあります。SQL Serverの変更追跡は後者に該当します。
ログベースのCDCであれば変更前データも取れますが、SQL Serverの変更追跡機能を使う場合、変更のあったPKの変更前のデータは取得できません。本記事ではSQL Serverの変更追跡機能を使ってBigQueryへ連携した変更のあったPKの変更ログに加えて、変更前ログを取得する方法をご紹介します。
techblog.zozo.com
datacater.io
BigQueryで実現する方法
BigQuery上で変更のあったPKの変更ログと変更前のログを取得する方法を紹介します。
変更のあった差分データを取得
SQL Serverの変更追跡機能で連携された差分テーブルから直近2日分の変更データを取得します。直近2日にしているのは、BigQueryパーティションでコストとパフォーマンスを向上させるのと、後ほど紹介する変更前のデータが変更のあった差分データに含まれていない場合に対応するためです。
streaming AS ( SELECT changetrack_type, changetrack_ver , bigquery_insert_time, <primary_key> AS primary_key, <columns> FROM <変更追跡で連携された差分テーブル> WHERE bigquery_insert_time >= TIMESTAMP_SUB(CAST(FORMAT_TIMESTAMP("%Y-%m-%d", TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 day), "Asia/Tokyo") AS timestamp), INTERVAL 9 HOUR)),
各カラムは以下の情報を含んでいます。
- changetrack_type: 変更処理。どのようは変更があったか確認できます(I: insert, U: update, D: delete)
- changetrack_ver：変更追跡のバージョン。SQL Serverのトランザクションごとに発行されるバージョンです。数値の大きいデータが最新になります
- primary_key: SQL ServerのPKをセットしています
- bigquery_insert_time: BigQueryに書き込まれた時刻.BigQueryのパーティション機能や遅延計測に使っています
- columns: 変更を追跡しているテーブルのカラム
最新の変更追跡バージョンを集計
先ほど抽出したデータを用いて、PKごとに最新の変更追跡バージョンを集計します。
streaming_latest_version AS ( SELECT primary_key, MAX(changetrack_ver) AS changetrack_ver_max FROM streaming -- set instead of tracking version WHERE bigquery_insert_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 60 second) GROUP BY primary_key ),
最新の変更追跡バージョンを用いて最新の変更ログを取得
PKごとに集計した最新の変更追跡バージョンと変更ログをJOINすることで、変更ログから変更のあったPKの最新のデータを取得できます。
streaming_latest AS ( SELECT changetrack_type, changetrack_ver, streaming.primary_key, <columns> FROM streaming INNER JOIN streaming_latest_version ON streaming.primary_key = streaming_latest_version.primary_key AND streaming.changetrack_ver = streaming_latest_version.changetrack_ver_max ),
変更前の変更追跡バージョンを取得
次に変更のあったPKの変更前のデータを取得します。変更のあった差分データと最新のバージョンをLEFT JOINし、突合できなかったデータ(changetrack_ver_max IS NULL)から変更前の変更追跡バージョンを取得します。変更のあった最新のバージョンは除外されるため変更前の変更追跡バージョンを取得できます。
streaming_before_latest_version AS ( SELECT primary_key, MAX(changetrack_ver) AS changetrack_ver_before_latest FROM ( SELECT streaming.primary_key, streaming.changetrack_ver, streaming_latest_version.changetrack_ver_max FROM streaming LEFT OUTER JOIN streaming_latest_version ON streaming.primary_key = streaming_latest_version.primary_key AND streaming.changetrack_ver = streaming_latest_version.changetrack_ver_max) WHERE changetrack_ver_max IS NULL GROUP BY primary_key ),
変更前の変更追跡バージョンを用いて変更前のログを取得
変更前の変更追跡バージョンを使って、変更のあったPKの変更前のデータを取得できます。
streaming_before_latest AS ( SELECT changetrack_type, changetrack_ver, streaming.primary_key, FROM streaming INNER JOIN streaming_before_latest_version ON streaming.primary_key = streaming_before_latest_version.changetrack_ver_before_latest AND streaming.changetrack_ver = streaming_before_latest_version.changetrack_ver_before_latest ),
前日分の全量テーブルから、差分データに含まれていない変更前データを取得
BigQueryのパーティションで絞りこんでいるため、変更のあった差分データの中に変更前のデータが含まれているとは限りません。変更のあった差分データの中に変更前のデータが含まれていない場合は前日分の全量テーブルから変更前のデータを取得します。
daily_before_latest AS ( SELECT CAST(NULL AS string) AS changetrack_type, CAST(NULL AS int64) AS changetrack_ver, <columns> FROM ( SELECT <primary_key> AS primary_key, <columns> FROM <前日の全量日付サフィックステーブル> WHERE _TABLE_SUFFIX IN (SUBSTR(FORMAT_TIMESTAMP("%Y%m%d", TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 day), "Asia/Tokyo"), 3))) WHERE primary_key NOT IN ( SELECT primary_key FROM streaming_before_latest_version) AND primary_key IN ( SELECT primary_key FROM streaming_latest_version) )
変更のあったPKの変更ログと変更前のログを取得
最後に変更データと変更前データをUNIONすることで、変更のあったPKの最新データと変更前のデータを取得できます。
SELECT * FROM streaming_latest UNION ALL SELECT * FROM streaming_before_latest UNION ALL SELECT * FROM daily_before_latest
BigQueryクエリ完成形
完成形のクエリは以下のようになります。
WITH # 本日分の差分テーブル streaming AS ( SELECT changetrack_type, changetrack_ver, bigquery_insert_time, <primary_key> AS primary_key, <columns> FROM <変更追跡で連携された差分テーブル> WHERE bigquery_insert_time >= TIMESTAMP_SUB(CAST(FORMAT_TIMESTAMP("%Y-%m-%d", TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 day), "Asia/Tokyo") AS timestamp), INTERVAL 9 HOUR)), # 差分テーブルから最新のバージョン streaming_latest_version AS ( SELECT primary_key, MAX(changetrack_ver) AS changetrack_ver_max FROM streaming -- set instead of tracking version WHERE bigquery_insert_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 60 second) GROUP BY primary_key ), # 差分テーブルから最新のログ streaming_latest AS ( SELECT changetrack_type, changetrack_ver, streaming.primary_key, <columns> FROM streaming INNER JOIN streaming_latest_version ON streaming.primary_key = streaming_latest_version.primary_key AND streaming.changetrack_ver = streaming_latest_version.changetrack_ver_max ), # 差分テーブルにある変更前のバージョン(最新の一つ前のバージョン) streaming_before_latest_version AS ( SELECT primary_key, MAX(changetrack_ver) AS changetrack_ver_before_latest FROM ( SELECT streaming.primary_key, streaming.changetrack_ver, streaming_latest_version.changetrack_ver_max FROM streaming LEFT OUTER JOIN streaming_latest_version ON streaming.primary_key = streaming_latest_version.primary_key AND streaming.changetrack_ver = streaming_latest_version.changetrack_ver_max) WHERE changetrack_ver_max IS NULL GROUP BY primary_key ), # 差分テーブルにある変更前のデータ(最新の一つ前のデータ) streaming_before_latest AS ( SELECT changetrack_type, changetrack_ver, streaming.primary_key, <columns> FROM streaming INNER JOIN streaming_before_latest_version ON streaming.primary_key = streaming_before_latest_version.changetrack_ver_before_latest AND streaming.changetrack_ver = streaming_before_latest_version.changetrack_ver_before_latest ), # 差分テーブルにない変更前のデータ daily_before_latest AS ( SELECT CAST(NULL AS string) AS changetrack_type, CAST(NULL AS int64) AS changetrack_ver, primary_key, <columns> FROM ( SELECT <primary_key> AS primary_key, <columns> FROM <前日の全量日付サフィックステーブル> WHERE _TABLE_SUFFIX IN (SUBSTR(FORMAT_TIMESTAMP("%Y%m%d", TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 day), "Asia/Tokyo"), 3))) WHERE primary_key NOT IN ( SELECT primary_key FROM streaming_before_latest_version) AND primary_key IN ( SELECT primary_key FROM streaming_latest_version) ) SELECT * FROM streaming_latest UNION ALL SELECT * FROM streaming_before_latest UNION ALL SELECT * FROM daily_before_latest
Dataflow JDBC テンプレート検証(Java)
Python版を調べてみたがクエリの上書きができなかったり、余計な通信が発生していたりと現時点で本番運用できる状態ではなかった。Java版が使えるか検証してみる。Java版はテンプレートが用意されていたので、PostgresとSQL Serverでそれぞれ検証してみた。結論としてはテンプレートをそのまま使うことはできないけど、クエリの上書きや無駄に実行環境からコネクション張る等Apache Beam Pythonで問題だった動きはしませんでした。本番運用でも使えそうです。
www.case-k.jp
www.case-k.jp
www.case-k.jp
テンプレートでサポートしてるのはBigQueryへの追加のみで全量置換はできていない。全量置換する場合自前で作る必要がある。パフォーマンス等調査は必要だがクエリの上書きやSQL Serverもサポートしていたので自前で用意すれば実務には使えそう。
.apply(
"Write to BigQuery",
BigQueryIO.writeTableRows()
.withoutValidation()
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
.withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory())
.to(options.getOutputTable()));
並列実行する場合JOBの同時実行数が懸念。プロジェクトごとに25。上限をあげてもオーガナイゼーションレベルで125が最大となっている。相談すれば増やせはするかも。
増やせないなら数百テーブルを高速かつ、並列実行でやる場合GKEのAutopilotで対応した方がコスト、パフォーマンス面で良さそうに思える。
- Each Google Cloud project can run at most 25 concurrent Dataflow jobs. - If you opt-in to organization level quotas, each organization can run at most 125 concurrent Dataflow jobs.
Note: If you would like to run more than 25 concurrent Dataflow jobs for your project or more than 125 concurrent Dataflow jobs for your organization, contact Google Cloud Support and we will increase the limit to a value that better suits your needs.
cloud.google.com
設定のパラメータ
- Postgres
# required parameters template: Jdbc to BigQuery jdbc connection URL string：jdbc:postgresql://<private ip address>:5432/beam jdbc driver class name：org.postgresql.Driver jdbc source sql query：<query> BigQuery output table : <project>:<dataset>.<table> gcs paths for Jdbc drivers： gs://<gcs-bucket>/postgresql-42.2.18.jre7.jar Temporary directory for BigQuery loading process:gs: gs://<gcs-bucket>//tmp/ # option parameters Jdbc connection username: <user-name> Jdbc connection password: <password> subnetwork: <subnetwork>
# required parameters jdbc connection URL string：jdbc:sqlserver://<private-ip>:1433;database=beam; jdbc driver class name：com.microsoft.sqlserver.jdbc.SQLServerDriver jdbc source sql query：<query> BigQuery output table : <project>:<dataset>.<table> gcs paths for Jdbc drivers： gs://<gcs-bucket>/mssql-jdbc-8.4.1.jre8.jar Temporary directory for BigQuery loading process:gs: gs://<gcs-bucket>//tmp/ # option parameters Jdbc connection username: <user-name> Jdbc connection password: <password> subnetwork: <subnetwork>
