メインコンテンツまでスキップ

LakeFlowジョブを使用して、バックフィルをサポートする定期的なクエリを設定する

一般的なシナリオは、ジョブによってオーケストレーションされ、定期的に実行されるクエリです。たとえば、毎日の終わりにクエリが実行され、その日のソース データセットの変更に基づいてシステムが更新されます。このチュートリアルでは、データをインポートする期間を特定するクエリを作成し、そのクエリを毎日実行するようにスケジュールするジョブを作成します。

このチュートリアルで作成されたクエリと問題はベスト プラクティスに一致しており、必要に応じて後でバックフィル ジョブを実行できるように設定されています。

前提条件​

このチュートリアルを完了するには、 Databricksのシステムテーブルにアクセスできる必要があります。

ステップ 1: クエリを作成する​

このチュートリアルでは、どのデータを取得するかを説明するために、 を使用するクエリを作成します。 たとえば、このチュートリアルでは、システム テーブルの請求データを使用して、毎日のDatabricksコストを計算します。

クエリでは次の 2 つの点が使用されます。

パラメーター

使用

data_interval_end

ジョブが実行される日付 (定期スケジュールの場合)。これは、処理される時間範囲の終了日です。または、バックフィル ジョブの場合は、バックフィルするデータの終了日。

lookback_days

何日分のデータをクエリするか。クエリはdata_interval_endから遡って検索します。これは通常、クエリが実行された時刻または日付であるため、前方検索ではなく後方検索を行う必要があります。

次のステップに従ってクエリを作成します。

  1. ワークスペースから、 をクリックします。プラスアイコン。 新しい 、それからノートブックのアイコン。 ノートブック 新しいノートブックを作成します。

  2. 名前のデフォルトはUntitled Notebook <date-time>です。ノートブックの上部にある名前をクリックし、 Query billing with parameters tutorialなどのわかりやすい名前を付けます。

  3. ノートブック エディターの上部にある言語セレクターから SQL を選択します。

  4. 最初のセルに次のコードを追加します。<catalog>と<schema>を、使用するアクセス権のあるカタログとスキーマに置き換えます。

    SQL
    USE CATALOG <catalog>;
    USE SCHEMA <schema>;

    CREATE TABLE IF NOT EXISTS tutorial_databricks_product_spend (billing_origin_product STRING, usage_date DATE, total_dollar_cost DECIMAL(12, 2));

    -- Process the last N days specified by :lookback_days ending on :data_interval_end
    INSERT INTO TABLE tutorial_databricks_product_spend
      REPLACE WHERE
        usage_date >= date_add(:data_interval_end, - CAST(:lookback_days AS INT)) AND usage_date < :data_interval_end
      SELECT
        usage.billing_origin_product,
        usage.usage_date,
        SUM(usage.usage_quantity * list_prices.pricing.effective_list.default) AS total_dollar_cost
      FROM
        system.billing.usage AS usage
          JOIN system.billing.list_prices AS list_prices
            ON usage.sku_name = list_prices.sku_name
            AND usage.usage_end_time >= list_prices.price_start_time
            AND (
              list_prices.price_end_time IS NULL
              OR usage.usage_end_time < list_prices.price_end_time
            )
      WHERE
        usage.usage_date >=
          date_add(:data_interval_end, -CAST(:lookback_days AS INT))
        AND usage.usage_date <
          :data_interval_end
      GROUP BY
        usage.billing_origin_product,
        usage.usage_date

  5. [編集] 、 [追加 ] をクリックして、2 つの問題を追加します。 問題には次の名前と値が含まれている必要があります。

名前

デフォルト値

lookback_days

1

data_interval_end

< なし >。この問題は常に必要です。

問題と、さまざまなタスク タイプまたはPythonからそれらにアクセスする方法については、 「タスクからの問題値へのアクセス」を参照してください。

クエリの準備が整いました。クエリはシステム テーブルから 1 日分のデータを読み取り、 REPLACE WHEREを使用して宛先テーブル内の既存のデータを置き換えます。 データを挿入するのではなく置き換えることにより、その日を 2 回目に実行してもクエリに悪影響はありません。実際、これにより、処理中にエラーが発生した場合やデータが遅れて到着した場合に、1 日を再実行することができます。

次のステップに従ってクエリをテストできます。

  1. ノートブックのセルの上にあるdata_interval_end値を、 yyyy-mm-ddの形式 (例: 2025-10-02 ) で指定します。
  2. 必要に応じて、円形アイコン。 接続し て使用するコンピュート リソースを選択します。
  3. クリック再生アイコン。 すべて実行します 。
  4. 実行が完了したら、カタログアイコン。左側のメニューから カタログを 選択し、クエリで設定したカタログとスキーマを選択します。

次に、クエリの定期的なスケジュールを作成します。

ステップ 2: ジョブを作成してクエリをスケジュールする​

  1. ワークスペースで、サイドバーの ワークフロー アイコン。 ジョブ & パイプライン をクリックします。

  2. [作成] をクリックし、 [ジョブ] を クリックします。 「タスク」 タブには、空の「タスク」ペインが表示されます。

注記

LakeFlowジョブ UI が オン の場合は、 ノートブック タイルをクリックして最初のタスクを設定します。 ノートブック タイルが利用できない場合は、 [別のタスク タイプを追加] をクリックし、 ノートブック を検索します。

  1. (オプション) ジョブ名 (デフォルトはNew Job <date-time> ) を自分のジョブ名に置き換えます。

  2. [タスク名] フィールドにタスクの名前を入力します。たとえば、 tutorial-databricks-spend 。

  3. 必要に応じて、 [タイプ] ドロップダウン メニューから [ノートブック] を選択します。

  4. [ソース] ドロップダウン メニューで [ワークスペース] を選択すると、以前に保存したノートブックを使用できるようになります。

  5. パス については、ファイル ブラウザーを使用して最初に作成したノートブックを見つけ、ノートブック名をクリックして、 確認 をクリックします。

  6. クリックプラスアイコン。 「問題」 セクションに 追加します 。 lookback_days問題を追加し、 値 を1にします。

  7. クリックプラスアイコン。 「問題」 セクションに 追加します 。 data_interval_end問題を追加します。 パラメータ化された値のリストを表示するには、 値の 横にある { } をクリックします。リストから{{job.trigger.time.iso_date}}を選択して値として挿入します。

    これにより、ジョブが実行がトリガーされた日付が保留として渡されます。

注記

1 日ではなく 1 時間など、より短い期間を振り返るクエリがある場合は、時間をつぶすために{{job.trigger.time.iso_datetime}}使用することもできます。この場合、クエリではどちらのオプションも機能しますが、 iso_date問題の意図を示しています。

  1. 「 タスクを作成 」をクリックします。

  2. タスクの右側にある詳細パネルの [スケジュールとトリガー] で、 [トリガーの追加] を クリックします。

  3. トリガータイプ で スケジュール を選択します。

  4. 1 日に 1 回実行されるアクティブなトリガーのデフォルトを維持します。

  5. 保存 をクリックします。

ジョブはクエリを毎日実行するようになりました。デフォルトでは、トリガーを作成した時刻と同じ時刻に実行されます。トリガーを編集し、 高度な トリガー タイプを選択して特定の時間を設定できます。

注記

このチュートリアル ジョブ実行を毎日行うことで料金が発生したくない場合は、停止するアイコン。作成したばかりのスケジュールで 停止します 。 これによりスケジュールは維持されますが、一時停止を解除するまで実行されません。いつでも手動で実行できます。

次に、バックフィルを実行して古いデータをテーブルにロードします。

ステップ 3: 古いデータのバックフィルを実行する​

バックフィルを実行して古いデータを入力できます。たとえば、先週のデータをテーブルに入力したい場合などです。次の手順では、過去 7 日間のデータをそれぞれ処理するために 7 回のバックフィル実行を作成します。

  1. ページ上部の 「今すぐ実行」 の横にある下矢印 ([下向き矢印アイコン]) をクリックします。

  2. 表示されるドロップダウンから 「バックフィルの実行」 を選択します。これにより 、バックフィルの実行 ダイアログが開きます。

  3. 日付範囲を、バックフィルする範囲に変更します。 開始を 7 日前の午前 12:00 に選択し、 終了 を今日の午前 12:00 に選択します。たとえば、開始時刻として09/14/2025, 12:00 AM選択し、終了時刻として09/21/2025, 12:00 AM選択できます。

  4. 各バックフィルの時間間隔として1 Day ごとを 選択します。

  5. [ジョブの問題] の下に、既存の問題がキーと値とともに表示されます。 data_interval_end問題が{{backfill.iso_datetime}}に設定され、 lookback_daysが1であることを確認します。

  6. 「実行」 をクリックしてバックフィル実行を開始します。 これにより、バックフィルの実行が 1 日ごとに 1 回ずつ、合計 7 回実行されます。

バックフィルは、ジョブの設定に応じて、並列または順次実行できます。バックフィルの詳細については、 「バックフィル ジョブ」を参照してください。

に最終更新