メインコンテンツまでスキップ
に最終更新

Lakeflow Jobsで初めてのワークフローを作成する

Lakeflow ジョブを使用して、サンプル データセットの読み取りと処理を行うタスクを調整します。このクイックスタートでは、次のことを行います。

  1. 新しいノートブックを作成し、物件予約のサンプルデータセットを読み込むコードを追加します。
  2. データセットを Unity Catalog に保存します。
  3. 新しいノートブックを作成し、Unity Catalog からデータセットを読み取り、年でフィルター処理して結果を表示するコードを追加します。
  4. 新しいジョブを作成し、ノートブックを使って2つのタスクを設定する。
  5. ジョブを実行し、結果を表示する。

必要条件​

ワークスペースが Unity Catalog に対応しており、 サーバレス ジョブ が有効になっている場合、デフォルトによって、サーバレス コンピュートでジョブが実行されます。 サーバレス コンピュートでジョブを実行するために、クラスター作成権限は必要ありません。

それ以外の場合は、ジョブ コンピュートを作成するための クラスター作成アクセス許可 、または汎用コンピュート リソースに対する アクセス許可 が必要です。

このクイックスタートは samples.wanderbricks.bookings から読み取ります。これはすべての Unity Catalog 有効化ワークスペースで利用できるため、設定するソースデータはありません。2番目のノートブックが読み取るテーブルに書き込むには、カタログ内にスキーマを作成する権限（USE CATALOG権限およびCREATE SCHEMA権限）が必要です。

これらの権限を設定するには、 Databricks管理者またはUnity Catalog権限リファレンスを参照してください。

ノートブックを作成する​

次の手順は、このワークフローで実行するノートブックを2つ作成します。

データの取得と保存​

サンプルデータセットを読み込み、それを Unity Catalog に保存するノートブックを作成するには：

  1. サイドバー 新しいアイコン「 新規」 をクリックしてから、「 ノートブック 」をクリックします。Databricks は、新しい空白のノートブックを作成して、デフォルト フォルダーに開きます。 デフォルト言語は、最後に使用した言語であり、ノートブックは、最後に使用したコンピュートリソースに自動的に添付されます。

  2. (オプション) ノートブックの名前を Retrieve booking data に変更します。

  3. 必要に応じて、 デフォルトの言語を Python に変更します。

  4. 次の Python コードをコピーして、ノートブックの最初のセルに貼り付けます。実行する前に、catalog と schema が書き込み可能な場所を指していることを確認してください。このコードは、スキーマが存在しない場合に作成します。

    Python
    catalog = "main"
    schema = "example_output"

    spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")

    bookings = spark.read.table("samples.wanderbricks.bookings")
    bookings.write.mode("overwrite").saveAsTable(f"{catalog}.{schema}.bookings")

フィルタリングされたデータの読み取りと表示​

データをフィルター処理して表示するノートブックを作成するには:

  1. サイドバー 新しいアイコン「 新規」 をクリックしてから、「 ノートブック 」をクリックします。

  2. （オプション）ノートブックの名前を Filter booking data に変更します。

  3. 以下の Python コードは、前のステップで保存したテーブルを読み取り、一時ビューを作成します。また、チェックイン年ごとにビュー内のデータをフィルタリングするために使用できるウィジェットも作成されます。最初のノートブックで使用したのと同じ catalog および schema 値を使用してください。

    Python
    from pyspark.sql.functions import year

    catalog = "main"
    schema = "example_output"

    bookings = spark.read.table(f"{catalog}.{schema}.bookings")
    bookings.createOrReplaceTempView("bookings_table")
    years = spark.sql("SELECT DISTINCT year(check_in) AS year FROM bookings_table").toPandas()["year"].tolist()
    years.sort()
    dbutils.widgets.dropdown("year", "2025", [str(x) for x in years])
    display(bookings.filter(year(bookings.check_in) == dbutils.widgets.get("year")))

ジョブを作成する​

作成するジョブは、2 つのタスクで構成されます。

最初のタスクを作成するには:

  1. ワークスペースで、サイドバーの ワークフローアイコン。 Jobs & パイプライン をクリックします。
  2. 「作成 」をクリックし、「 ジョブ」 をクリックします。
  3. 最初のタスクを構成するには、 ノートブック タイルをクリックします。 ノートブック タイルが利用できない場合は、 [別のタスク タイプを追加] をクリックし、 ノートブック を検索します。
  4. (オプション)ジョブの名前 (デフォルトは New Job <date-time> ) をジョブ名に置き換えます。
  5. タスク名 フィールドにタスクの名前を入力します。（例： retrieve-bookings ）
  6. 必要に応じて、[ タイプ ] ドロップダウン メニューから [ノートブック ] を選択します。
  7. [ ソース ] ドロップダウン メニューで [ ワークスペース] を選択すると、以前に保存したノートブックを使用できます。
  8. [パス] で、ファイル ブラウザーを使用して最初に作成したノートブックを検索し、ノートブック名をクリックして [確認] をクリックします。
  9. タスクの保存 をクリックします。画面の右上隅に通知が表示されます。

2 番目のタスクを作成するには:

  1. プラスアイコン。 タスクの追加 > ノートブック をクリックします。
  2. タスク名 フィールドにタスクの名前を入力します。（例： filter-bookings ）
  3. [パス] で、ファイル ブラウザーを使用して作成した 2 番目のノートブックを見つけ、ノートブック名をクリックして [確認] をクリックします。
  4. パラメーター の下の 追加 をクリックします。 キー フィールドにyearを入力します。 値 フィールドに2025を入力します。
  5. タスクの保存 をクリックします。

ジョブを実行する​

ジョブをすぐに実行するには、右上隅の 「今すぐ実行」ボタン をクリックします。

実行の詳細を表示する​

  1. 実行 タブをクリックし 、[ 開始時刻] 列 のリンクをクリックして、表示する実行を開きます。

  2. いずれかのタスクをクリックすると、出力と詳細が表示されます。たとえば、 filter-bookings タスクをクリックすると、フィルタータスクの出力とランの詳細が表示されます：

    フィルタリングされた予約結果を表示

異なるパラメーターでの実行​

ジョブを再実行し、別の年の予約をフィルタリングするには、次の手順を実行します：

  1. ブルーダウンキャレット [今すぐ実行] の横にある をクリックし、[ 別の設定で今すぐ実行] を選択します。
  2. 値 フィールドに2024を入力します。
  3. 実行 をクリックします。

その他のリソース​

このページの見出し