Jobs プラグイン

AppKit アプリケーションから Databricks Lakeflow Jobs をトリガーし、監視します。

主な機能:

• 名前付きjobキーによる複数jobのサポート
• 環境変数からのjobの自動検出
• SSE ストリーミングでステータスを更新する run-and-wait
• Zod スキーマによるパラメータの検証
• タスクタイプに応じたパラメータのマッピング (notebook、python_wheel、sql など)

基本的な使い方

import { createApp, server, jobs } from "@databricks/appkit";

await createApp({
  plugins: [server(), jobs()],
});

明示的な jobs 設定がない場合、pluginは環境変数 DATABRICKS_JOB_ID を読み取り、default キーとして登録します。

設定オプション

job単位の設定 (JobConfig)

環境変数

単一 job モード

DATABRICKS_JOB_ID を設定すると、default キー配下に job を 1 つ公開します:

DATABRICKS_JOB_ID=123456

const handle = AppKit.jobs("default");

マルチjobモード

job ごとに DATABRICKS_JOB_<NAME> を設定します:

DATABRICKS_JOB_ETL=123456
DATABRICKS_JOB_ML=789012

const etl = AppKit.jobs("etl");
const ml  = AppKit.jobs("ml");

環境変数名は大文字、jobキーは小文字になります。環境から検出されたjobは、明示的な jobs 設定とマージされ、明示的な設定が優先されます。

パラメータの検証

params を使うと、実行時に Zod スキーマを強制できます。不正なパラメータは、job がトリガーされる前に 400 で拒否されます。

import { z } from "zod";

jobs({
  jobs: {
    etl: {
      params: z.object({
        startDate: z.string(),
        endDate: z.string(),
        dryRun: z.boolean().optional(),
      }),
    },
  },
})

タスクタイプのマッピング

taskType を設定すると、plugin は検証済みのパラメータを適切な SDK リクエストフィールドへ自動的にマッピングします。

jobs({
  jobs: {
    etl: {
      taskType: "notebook",
      params: z.object({
        startDate: z.string(),
        endDate: z.string(),
      }),
    },
  },
})

taskType を省略した場合、パラメータはそのまま SDK に渡されます。

実行コンテキスト

jobは常にアプリのサービスプリンシパルとして実行されます。アプリのリソースバインディング (databricks.yml) でサービスプリンシパルに CAN_MANAGE_RUN が付与されるため、ユーザーは個別に権限を付与されなくてもrunを開始できます。Jobs UI のrunごとの実行者表示は、実際のユーザーではなくアプリのサービスプリンシパルになります。

HTTP endpoint

すべてのルートは /api/jobs 配下にマウントされます。

run をトリガーする

{ "runId": 12345 } を返します。

?stream=true を追加すると、run が完了するまでポーリングを行い、SSE でステータス更新を受信します:

各 SSE イベントには { status, timestamp, run } が含まれます。

run の一覧取得

{ "runs": [...] } を返します。limit は 1〜100 の範囲に丸められ、デフォルトは 20 です。

run の詳細を取得する

最新のステータスを取得

最新の run について { "status": "TERMINATED", "run": { ... } } を返します。

run をキャンセルする

成功時は 204 No Content を返します。

プログラムによるアクセス

この plugin は、キーを指定して job を選択する呼び出し可能オブジェクトをエクスポートします:

const AppKit = await createApp({
  plugins: [
    server(),
    jobs({
      jobs: {
        etl: { taskType: "notebook" },
      },
    }),
  ],
});

const etl = AppKit.jobs("etl");

// runをトリガーする
const result = await etl.runNow({ startDate: "2025-01-01" });
if (result.ok) {
  console.log("Run ID:", result.data.run_id);
}

// runをトリガーして完了までポーリングする
for await (const status of etl.runAndWait({ startDate: "2025-01-01" })) {
  console.log(status.status); // "PENDING"、"RUNNING"、"TERMINATED" など
}

// 読み取り操作
await etl.lastRun();
await etl.listRuns({ limit: 10 });
await etl.getRun(12345);
await etl.getRunOutput(12345);
await etl.getJob();

// キャンセル
await etl.cancelRun(12345);

すべてのメソッドは ExecutionResult<T> を返します。result.data にアクセスする前に result.ok を確認してください。

実行のデフォルト値