Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

実行をスケジュールする

このドキュメントでは、Dataform で次の操作を行う方法について説明します。

ワークフロー構成で実行をスケジュールする。
Workflows と Cloud Scheduler で実行をスケジュールする。
Managed Service for Apache Airflow で実行をスケジュールする。
Cloud Build トリガーを使用して実行を自動化する。

次の表は、各メソッドを比較したものです。

メソッド	目標	トリガーのタイプ
ワークフロー構成	Dataform 内の標準の時間ベースの本番環境実行。	時間ベース
Workflows と Cloud Scheduler	他の API とともに Dataform を軽量でサーバーレスにオーケストレートします。	時間ベースまたはイベントドリブン
マネージド Airflow	BigQuery の外部に依存関係がある複雑なパイプライン。	時間ベースまたはイベントドリブン
Cloud Build トリガー	Git push の直後にリリースを自動化する。	イベントドリブン

始める前に

ワークフロー構成で実行をスケジュールするか、ワークフローと Cloud Scheduler で実行をスケジュールするには、次の操作を行います。

Google Cloud コンソールで、[Dataform] ページに移動します。

Dataform に移動
リポジトリを作成または選択します。

注: Dataform リポジトリのリストには、Dataform ワークフロー開発用に作成されたリポジトリのみが含まれます。ノートブックや保存済みクエリなどの BigQuery Studio アセットのリポジトリは個別に管理され、このリストには表示されません。
リリース構成を作成する

Managed Service for Apache Airflow で実行をスケジュールするには、次の操作を行います。

必要なロール

このドキュメントのタスクを完了するために必要な権限を取得するには、管理者に次の IAM ロールを付与するよう依頼してください。

リポジトリに対する Dataform 管理者（roles/dataform.admin）
カスタムサービスアカウントに対するサービスアカウントユーザー（roles/iam.serviceAccountUser）
Managed Airflow 環境のサービスアカウントに対する Composer ワーカー（roles/composer.worker）
Cloud Build を使用して実行を自動化する:
- カスタムサービスアカウントに対するサービスアカウント管理者（roles/iam.serviceAccountAdmin）
- プロジェクトに対する Cloud Build 編集者（roles/cloudbuild.builds.editor）

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

必要な権限は、カスタムロールや他の事前定義ロールから取得することもできます。

ワークフロー構成の作成時にカスタムサービスアカウントを使用するには、カスタムサービスアカウントへのアクセス権を付与します。

ワークフロー構成の作成時に Google アカウントのユーザー認証情報を使用するには（プレビュー）、Google アカウントへのアクセス権を付与します。

ワークフロー構成のスケジュール設定された実行を有効にするには、ワークフロー構成で使用されるカスタムサービスアカウントのデフォルトの Dataform サービスエージェントに iam.serviceAccounts.actAs 権限を付与する必要があります。この権限は、サービスアカウントユーザーロール（roles/iam.serviceAccountUser）で使用できます。詳細については、厳格な act-as モードを使用するをご覧ください。

スケジュールのセキュリティを強化するには、拡張されたスケジュール設定の権限を実装するをご覧ください。

ワークフロー構成で実行をスケジュールする

このセクションでは、Dataform でワークフロー構成を作成して、ワークフローの実行をスケジュールし、構成する方法について説明します。ワークフロー構成を使用して Dataform ワークフローをスケジュールで実行する方法について確認できます。

ワークフロー構成について

BigQuery でワークフローアクションのすべてまたは選択した Dataform の実行をスケジュールするには、ワークフロー構成を作成します。ワークフロー構成では、コンパイルリリース構成を選択し、実行するワークフローアクションを選択して、実行スケジュールを設定します。

その後、Dataform は、ワークフロー構成のスケジュールされた実行時に、リリース構成の最新のコンパイル結果から選択したアクションを BigQuery にデプロイします。Dataform API workflowConfigs を使用して、ワークフロー構成の実行を手動でトリガーすることもできます。

Dataform ワークフロー構成には、次の実行設定が含まれています。

ワークフロー構成の ID。
リリース構成。
サービスアカウント。

これは、ワークフロー構成に関連付けられているカスタムサービスアカウントです。 Google Cloud プロジェクトに関連付けられているカスタムサービスアカウントを選択することも、別のサービスアカウントを手動で入力することもできます。デフォルトでは、ワークフロー構成はリポジトリと同じサービスアカウントを使用します。

サービスアカウントの認証情報は、スケジュールされたワークフロー構成の作成と実行のデフォルトの認可方法です。

注: リポジトリ内のすべてのワークフロー構成に、最大 100 個の一意のサービスアカウントを関連付けることができます。
Google アカウントのユーザー認証情報（プレビュー）

Google アカウントのユーザー認証情報は、手動のスケジュールされていないワークフロー構成の作成と実行のデフォルトの認証方法です。詳しくは、Google アカウントを認可するをご覧ください。
実行するワークフローアクション:
- すべてのアクション。
- アクションの選択。
- タグの選択。
実行スケジュールとタイムゾーン。

ワークフロー構成を作成する

Dataform ワークフロー構成を作成する手順は次のとおりです。

リポジトリで、[リリースとスケジュール] に移動します。
[ワークフロー構成] セクションで、[作成] をクリックします。
[ワークフロー構成を作成] ペインの [構成 ID] フィールドに、ワークフロー構成の一意の ID を入力します。

ID には数字、英字、ハイフン、アンダースコアのみを使用できます。
[リリース構成] メニューで、コンパイルリリース構成を選択します。
[認証] セクションで、Google アカウントのユーザー認証情報またはサービスアカウントを使用してワークフロー構成を認可します。
- Google アカウントのユーザー認証情報（プレビュー）を使用するには、[自分のユーザー認証情報で実行] を選択します。
  
  省略可: [拡張アクセスオプション] セクションで、ワークフローに必要な追加サービスを選択します。
  - Knowledge Catalog: Google Cloud Knowledge Catalog メタデータの更新を許可します。
  - Google ドライブ: Google ドライブファイルへの読み取り専用アクセスを許可します。
  - Bigtable: Google Bigtable データへの読み取り専用アクセスを許可します。
- カスタムサービスアカウントを使用するには、[選択したサービスアカウントで実行] を選択し、アクセス権のある Google Cloud プロジェクトに関連付けられているサービスアカウントを選択します。サービスアカウントを選択しない場合、ワークフロー構成ではリポジトリのサービスアカウントが使用されます。
注: デフォルトの Dataform サービスエージェントを使用してワークフローを実行することはできません。カスタムサービスアカウントまたは Google アカウントのユーザー認証情報のいずれかを使用する必要があります。
省略可: [スケジュール頻度] フィールドに、unix-cron 形式で実行の頻度を入力します。

Dataform が対応するリリース構成で最新のコンパイル結果を確実に実行するには、コンパイル結果の作成時刻とスケジュールされた実行時刻の間に少なくとも 1 時間の間隔を設けます。
省略可: [タイムゾーン] メニューで、実行のタイムゾーンを選択します。

デフォルトのタイムゾーンは UTC です。
実行するワークフローアクションを選択します。
- ワークフロー全体を実行するには、[すべてのアクション] をクリックします。
- ワークフローで選択したアクションを実行するには、[アクションの選択] をクリックして、アクションを選択します。
- 選択したタグを持つアクションを実行するには、[タグの選択] をクリックしてタグを選択します。
- 省略可: 選択したアクションまたはタグとそれらの依存関係を実行するには、[依存関係を含める] オプションを選択します。
- 省略可: 選択したアクションまたはタグとそれらの依存関係を実行するには、[依存者を含める] オプションを選択します。
- 省略可: すべてのテーブルをゼロから再構築するには、[フル更新で実行] オプションを選択します。
  
  このオプションを使用しない場合、Dataform は、増分テーブルをゼロから再構築せずに更新します。
- 省略可: 優先度の高いインタラクティブジョブとして実行する（デフォルト）オプションを使用して、BigQuery クエリジョブの優先度を設定します。デフォルトでは、BigQuery はクエリをインタラクティブクエリジョブとして実行します。このジョブは、できるだけ早く実行を開始することを目的としています。このオプションをオフにすると、クエリは優先度の低いバッチクエリジョブとして実行されます。
[作成] をクリックします。認証方法に [自分のユーザー認証情報で実行] を選択した場合は、Google アカウントを認可する必要があります（プレビュー）。

たとえば、次のワークフロー構成では、CEST タイムゾーンで 1 時間ごとに hourly タグを使用してアクションを実行します。

構成 ID：production-hourly
リリース構成：
頻度：0 * * * *
タイムゾーン：Central European Summer Time (CEST)
ワークフローアクションの選択: タグの選択、hourly タグ

Google アカウントを認可する

Google アカウントのユーザー認証情報でリソースを認証するには、BigQuery パイプラインが Google アカウントのアクセストークンを取得し、ユーザーに代わってソースデータにアクセスするための権限を手動で付与する必要があります。OAuth ダイアログインターフェースで、手動で承認できます。拡張アクセスオプションを選択した場合は、Google ドライブや Knowledge Catalog などのサービスへのアクセス権を付与する必要があります。

BigQuery パイプラインに権限を付与する必要があるのは 1 回のみです。

付与した権限を取り消すには、次の手順に沿って操作します。

Google アカウントページにアクセスします。
[BigQuery パイプライン] をクリックします。
[アクセス権を削除] をクリックします。

新しい Google アカウントの所有者が以前にワークフロー構成を作成したことがない場合は、認証情報を更新してワークフロー構成のオーナーを変更するには手動で承認する必要もあります。

ワークフロー構成を編集する

ワークフロー構成を編集するには、次の手順を行います。

リポジトリで、[リリースとスケジュール] に移動します。
編集するワークフロー構成で、 [その他] メニューをクリックし、[編集] をクリックします。
[ワークフロー構成を編集] ペインで、ワークフロー構成の設定を編集し、[保存] をクリックします。

ワークフロー構成を削除する

ワークフロー構成を作成するには、次の手順を行います。

リポジトリで、[リリースとスケジュール] に移動します。
削除するワークフロー構成で、 [その他] メニューをクリックし、[削除] をクリックします。
[リリース構成の削除] ダイアログで、[削除] をクリックします。

Workflows と Cloud Scheduler で実行をスケジュールする

このセクションでは、Workflows と Cloud Scheduler を使用して Dataform ワークフローの実行をスケジュールする方法について説明します。

スケジュールされたワークフローの実行について

Dataform ワークフローの実行頻度を設定するには、Workflows ワークフローをトリガーする Cloud Scheduler ジョブを作成します。Workflows は、定義したオーケストレーションワークフローでサービスを実行します。

Workflows は、Dataform ワークフローを 2 段階のプロセスで実行します。まず、Git プロバイダから Dataform リポジトリコードを pull してコンパイルし、コンパイル結果を導きます。次に、コンパイル結果を使用して Dataform ワークフローを作成し、設定した頻度で実行します。

スケジュール設定されたオーケストレーションワークフローを作成する

Dataform ワークフローの実行をスケジュールするには、Workflows を使用してオーケストレーションワークフローを作成し、Cloud Scheduler ジョブをトリガーとして追加します。

Workflows はサービスアカウントを使用して、ワークフローがGoogle Cloud リソースにアクセスできるようにします。サービスアカウントを作成し、次の権限を付与します。
- Dataform 編集者のロール（roles/dataform.editor）。
- Dataform で使用されるカスタムサービスアカウントに対するサービスアカウントユーザーのロール（roles/iam.serviceAccountUser）。
- オーケストレーションワークフローの管理に必要な最小限の権限。詳細については、 Google Cloud リソースにアクセスする権限をワークフローに付与するをご覧ください。

オーケストレーションワークフローを作成し、次の YAML ソースコードをワークフロー定義として使用します。

main:
    steps:
    - init:
        assign:
        - repository: projects/PROJECT_ID/locations/REPOSITORY_LOCATION/repositories/REPOSITORY_ID
    - createCompilationResult:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1/" + repository + "/compilationResults"}
            auth:
                type: OAuth2
            body:
                gitCommitish: GIT_COMMITISH
        result: compilationResult
    - createWorkflowInvocation:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1/" + repository + "/workflowInvocations"}
            auth:
                type: OAuth2
            body:
                compilationResult: ${compilationResult.body.name}
        result: workflowInvocation
    - complete:
        return: ${workflowInvocation.body.name}

次のように置き換えます。

PROJECT_ID: 実際の Google Cloud プロジェクト ID。
REPOSITORY_LOCATION: Dataform リポジトリのロケーション。
REPOSITORY_ID: Dataform リポジトリの名前。
GIT_COMMITISH: Dataform コードを実行する Git ブランチ。新しく作成したリポジトリの場合は、main に置き換えます。

Cloud Scheduler を使用してオーケストレーションワークフローをスケジュールします。

Dataform ワークフローのコンパイル結果作成リクエストをカスタマイズする

既存のオーケストレーションワークフローを更新し、Dataform ワークフローのコンパイル結果作成リクエスト設定を YAML 形式で定義できます。設定の詳細については、projects.locations.repositories.compilationResults REST リソースリファレンスをご覧ください。

たとえば、コンパイル時にすべての操作に _dev schemaSuffix 設定を追加するには、createCompilationResult ステップの本文を次のコードスニペットに置き換えます。

    - createCompilationResult:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1/" + repository + "/compilationResults"}
            auth:
                type: OAuth2
            body:
                gitCommitish: GIT_COMMITISH
                codeCompilationConfig:
                    schemaSuffix: dev

Workflows 実行リクエストで追加設定をランタイム引数として渡し、変数を使用してこれらの引数にアクセスすることもできます。詳細については、実行リクエストでランタイム引数を渡すをご覧ください。

Dataform ワークフロー呼び出しリクエストをカスタマイズする

既存のオーケストレーションワークフローを更新し、Dataform ワークフロー呼び出しリクエストの設定を YAML 形式で定義できます。呼び出しリクエストの設定の詳細については、projects.locations.repositories.workflowInvocations REST リソースリファレンスをご覧ください。

たとえば、hourly タグが付いたアクションのみをすべての推移的依存関係を含めて実行するには、createWorkflowInvocation の本文を次のコードスニペットに置き換えます。

    - createWorkflowInvocation:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1/" + repository + "/workflowInvocations"}
            auth:
                type: OAuth2
            body:
                compilationResult: ${compilationResult.body.name}
                invocationConfig:
                    includedTags:
                    - hourly
                    transitiveDependenciesIncluded: true

Managed Airflow で実行をスケジュールする

Managed Airflow 2 を使用して、Dataform の実行をスケジュールできます。Dataform は Managed Airflow 1 をサポートしていません。

Managed Airflow 2 で Dataform の実行スケジュールを管理するには、Airflow 有向非巡回グラフ（DAG）で Dataform オペレーターを使用します。Dataform ワークフローの呼び出しをスケジュールする Airflow DAG を作成できます。

Dataform は、さまざまな Airflow 演算子を提供します。これには、コンパイル結果の取得、ワークフロー呼び出しの取得、ワークフロー呼び出しのキャンセルを行う演算子が含まれます。使用可能な Dataform Airflow オペレーターの完全なリストについては、Google Dataform 演算子をご覧ください。

`google-cloud-dataform` PyPI パッケージをインストールする

Managed Airflow 2 バージョン 2.0.25 以降を使用している場合、このパッケージは環境にプリインストールされています。インストールする必要はありません。

以前のバージョンの Managed Airflow 2 を使用している場合は、google-cloud-dataform PyPI パッケージをインストールします。

[PyPI パッケージ] セクションで、バージョン ==0.2.0 を指定します。

Dataform ワークフローの呼び出しをスケジュールする Airflow DAG を作成する

Managed Airflow 2 で Dataform ワークフローのスケジュール設定された実行を管理するには、Dataform Airflow オペレーターを使用して DAG を作成し、環境のバケットにアップロードします。

次のコードサンプルは、Dataform コンパイル結果を作成して Dataform ワークフローの呼び出しを開始する Airflow DAG を示しています。

from datetime import datetime

from airflow import models
from airflow.models.baseoperator import chain
from airflow.providers.google.cloud.operators.dataform import (
    DataformCreateCompilationResultOperator,
    DataformCreateWorkflowInvocationOperator,
)

DAG_ID = "dataform"
PROJECT_ID = "PROJECT_ID"
REPOSITORY_ID = "REPOSITORY_ID"
REGION = "REGION"
GIT_COMMITISH = "GIT_COMMITISH"

with models.DAG(
    DAG_ID,
    schedule_interval='@once',  # Override to match your needs
    start_date=datetime(2022, 1, 1),
    catchup=False,  # Override to match your needs
    tags=['dataform'],
) as dag:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": GIT_COMMITISH,
        },
    )
    create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
        task_id='create_workflow_invocation',
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
         workflow_invocation={
            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}"
        },
    )


create_compilation_result >> create_workflow_invocation