Cloud Composer を使用するワークフロー

このドキュメントでは、課金対象である次の Google Cloudコンポーネントを使用します。

Managed Service for Apache Spark
Compute Engine
Cloud Composer

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを生成できます。

新規の Google Cloud ユーザーは無料トライアルをご利用いただける場合があります。

始める前に

プロジェクトを設定する

Google Cloud アカウントにログインします。 Google Cloudを初めて使用する場合は、アカウントを作成して、実際のシナリオでの Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分を差し上げます。

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, and Cloud Composer APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Google Cloud CLI をインストールします。

外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

gcloud CLI を初期化するには、次のコマンドを実行します。

gcloud init

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, and Cloud Composer APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Google Cloud CLI をインストールします。

外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

gcloud CLI を初期化するには、次のコマンドを実行します。

gcloud init

Managed Service for Apache Spark ワークフローテンプレートを作成する

ローカルターミナルウィンドウまたは Cloud Shell に次のコマンドをコピーして実行し、ワークフローテンプレートを作成して定義します。

sparkpi ワークフローテンプレートを作成します。

gcloud dataproc workflow-templates create sparkpi \
    --region=us-central1

Spark ジョブを sparkpi ワークフローテンプレートに追加します。「compute」step-id フラグは、SparkPi ジョブを示します。

gcloud dataproc workflow-templates add-job spark \
    --workflow-template=sparkpi \
    --step-id=compute \
    --class=org.apache.spark.examples.SparkPi \
    --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
    --region=us-central1 \
    -- 1000

ワークフローを実行するには、マネージドの単一ノードクラスタを使用します。Managed Service for Apache Spark によってクラスタが作成され、ワークフローがクラスタ上で実行され、ワークフローの完了時にクラスタが削除されます。
```
gcloud dataproc workflow-templates set-managed-cluster sparkpi \
    --cluster-name=sparkpi \
    --single-node \
    --region=us-central1
      
```
ワークフローテンプレートの作成を確認します。
コンソール

Google Cloud コンソールの Managed Service for Apache Spark の [ワークフロー] ページで sparkpi 名をクリックし、[ワークフローテンプレートの詳細] ページを開きます。ワークフローテンプレートの名前をクリックして、sparkpi テンプレート属性を確認します。
gcloud コマンド

次のコマンドを実行します。
```
gcloud dataproc workflow-templates describe sparkpi --region=us-central1
    
```

DAG を作成して Cloud Storage にアップロードする

Cloud Composer 環境を作成するか、既存の環境を使用します。
環境変数を設定します。
Airflow UI
1. ツールバーで、[Admin] > [Variables] をクリックします。
2. [Create] をクリックします。
3. 次の情報を入力します。
  - Key: project_id
  - Val: PROJECT_ID - Google Cloud プロジェクト ID
4. [Save] をクリックします。
gcloud コマンド

次のコマンドを入力します。
- ENVIRONMENT は、Cloud Composer 環境の名前です。
- LOCATION は、Cloud Composer 環境が配置されているリージョンです。
- PROJECT_ID は、Cloud Composer 環境を含むプロジェクトのプロジェクト ID です。
```
    gcloud composer environments run ENVIRONMENT --location LOCATION variables set -- project_id PROJECT_ID
    
```

次の DAG コードを、ローカルで「composer-dataproc-dag.py」という名前のファイルにコピーします。このコードは、DataprocWorkflowTemplateInstantiateOperator を使用します。

Airflow 2


"""Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
Spark Pi Job.

This DAG relies on an Airflow variable
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocInstantiateWorkflowTemplateOperator,
)
from airflow.utils.dates import days_ago

project_id = "{{var.value.project_id}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": project_id,
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "dataproc_workflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = DataprocInstantiateWorkflowTemplateOperator(
        # The task id of your job
        task_id="dataproc_workflow_dag",
        # The template id of your workflow
        template_id="sparkpi",
        project_id=project_id,
        # The region for the template
        region="us-central1",
    )

Airflow 1


"""Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
Spark Pi Job.

This DAG relies on an Airflow variable
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
"""

import datetime

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils.dates import days_ago

project_id = "{{var.value.project_id}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": project_id,
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "dataproc_workflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
        # The task id of your job
        task_id="dataproc_workflow_dag",
        # The template id of your workflow
        template_id="sparkpi",
        project_id=project_id,
        # The region for the template
        # For more info on regions where Dataflow is available see:
        # https://cloud.google.com/dataflow/docs/resources/locations
        region="us-central1",
    )

Cloud Storage の環境フォルダに DAG をアップロードします。アップロードが正常に完了したら、Cloud Composer 環境のページにある DAG フォルダのリンクをクリックします。

タスクのステータスを表示する

Airflow UI

Airflow ウェブインターフェースを開きます。
DAG ページで、DAG 名（例: dataproc_workflow_dag）をクリックします。
DAG の詳細ページで、[Graph View] をクリックします。
ステータスを確認します。
- Failed: タスクの周囲に赤色のボックスが表示されます。ポインタをタスクの上に置くと、[State: Failed] と表示されます。
- Success: タスクの周囲に緑色のボックスが表示されます。タスクにカーソルを合わせて [State: Success] を確認することもできます。

コンソール

[ワークフロー] タブをクリックして、ワークフローのステータスを確認します。

gcloud コマンド

gcloud dataproc operations list \
    --region=us-central1 \
    --filter="labels.goog-dataproc-workflow-template-id=sparkpi"

クリーンアップ

Google Cloud アカウントに課金されないようにするには、このチュートリアルで使用したリソースを削除します。

Cloud Composer 環境を削除します。
ワークフローテンプレートを削除します。

次のステップ

Managed Service for Apache Spark ワークフローテンプレートの概要をご覧ください。
ワークフロースケジューリングソリューションを確認する

Cloud Composer を使用するワークフロー コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

始める前に

プロジェクトを設定する

Managed Service for Apache Spark ワークフロー テンプレートを作成する

コンソール

gcloud コマンド

DAG を作成して Cloud Storage にアップロードする

Airflow UI

gcloud コマンド

Airflow 2

Airflow 1

タスクのステータスを表示する

Airflow UI

コンソール

gcloud コマンド

クリーンアップ

次のステップ

Cloud Composer を使用するワークフロー

Managed Service for Apache Spark ワークフローテンプレートを作成する