grasys

DagsterをCloud Runで構築してみる

DagsterをCloud Runで構築してみる

DagsterをCloud Runで構築してみる

はじめまして！エンジニアのUemaです。

最近、試しでデータオーケストレーションツールのDagsterを触っていて、実際に運用していく事になった場合にどのような構成が良いのか考えていました。

フルマネージドでコンテナをデプロイできる点、Cloud Runサービスはゼロインスタンスにでき利用した分だけ課金が発生する点からCloud Runが良いなと思いました。私はよく利用もしてます。

また、Cloud Runにワーカープールという継続的なバックグラウンド処理をしてくれる仲間も登場しています。(Cloud Runワーカープールは稼働時間中課金されます)

Cloud Runを使用して構築していくことを決めました！

大好きCloud Run！！

Dagster & Cloud Runの構成

Dagster WebserverはCloud Runサービス、Dagster DaemonはCloud Runワーカープール、Dagster RunWorkerはCloud Runジョブを使用して構築します。

今回の構成では構築が複雑になってしまうため、WebserverやDaemonに対してコードの共有をするための独立したCode Location Serverは構築しないです。

Dagsterの管理や履歴を保存するためのDBが必要なため、Cloud SQL for PostgreSQLを使用します。

  • 構成図は下記の通り

    DagsterとCloudRunの構成図

構築

上記の構成図を元に構築していきます。

ライブラリのインストール

Cloud Runジョブを実行できるように dagster-contrib-gcp の CloudRunRunLauncher を使用するため、インストールしておきます。

dagster-contrib-gcp はコミュニティ提供なので感謝して使います。

他にライブラリが必要な場合はインストールしてください。

Bash
# uvを使用
uv add dagster dagster-webserver dagster-postgres dagster-contrib-gcp

コンテナ内ディレクトリ構成

コンテナ内のディレクトリ構成は下記のような感じ

Plain text
dagster/
|-- app
|   |-- pyproject.toml
|   |-- src/
|   |   `-- my_project/
|   |       |-- __init__.py
|   |       |-- assets/
|   |       |   |-- __init__.py
|   |       |   |-- lake.py
|   |       |   |-- mart.py
|   |       |   `-- warehouse.py
|   |       |-- definitions.py
|   |       |-- jobs.py
|   |       |-- resources/
|   |       |   |-- __init__.py
|   |       |   `-- bigquery.py
|   |       |-- schedules.py
|   |       `-- sensors.py
|   |-- uv.lock
|   `-- workspace.yaml
`-- dagster_home/
    `-- dagster.yaml

workspace.yaml

Dagsterのコードの場所を定義・設定します。

YAML
load_from:
  #.....
  - python_module:
      module_name: my_project.definitions # src以下のmy_project/definitions.pyを設定
      location_name: my_data_platform
  #.....

dagster.yaml

Dagsterのランチャーやストレージなどの設定をします。

YAML
#.....

scheduler:
  module: dagster.core.scheduler
  class: DagsterDaemonScheduler

run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 5

run_launcher:
  module: dagster_contrib_gcp.cloud_run.run_launcher
  class: CloudRunRunLauncher
  config:
    project:
      env: GOOGLE_CLOUD_PROJECT
    region:
      env: GOOGLE_CLOUD_REGION
    # cloudbuild.yaml の Cloud RunのJob名 と一致させること。
    job_name_by_code_location:
      my_data_platform: dagster-run-worker

storage:
  postgres:
    postgres_db:
      username:
        env: DAGSTER_POSTGRES_USER
      password:
        env: DAGSTER_POSTGRES_PASSWORD
      # Cloud SQL Unix ソケットのパスは libpq 標準の PGHOST から解決するため、hostnameは空にする
      hostname: ""
      db_name:
        env: DAGSTER_POSTGRES_DB
      port: 5432

#.....

cloudbuild.yaml

Cloud Buildを使ってデプロイするので、cloudbuild.yamlを作成します。

サービスアカウントは必要に応じて権限を追加してください。

  • Run Worker用のサービスアカウントの権限には、Cloud SQL クライアント、BigQuery、GCSなど実処理に必要な権限
  • Daemon用のサービスアカウントの権限には、Cloud SQL クライアント、Secret Manager Secret Accessor、Cloud Run デベロッパー など
  • Webserver用のサービスアカウントの権限には、Cloud SQL クライアント、Secret Manager Secret Accessor など
YAML
# 前提:
# - Artifact Registry、Cloud SQL for PostgreSQL、Secret Manager の各リソースは作成済み
steps:
  #.....

  # Dagsterのバージョンをアップグレードする際には、マイグレーションが必要になる場合があるため、マイグレーション用のJobをデプロイ
  - id: deploy-migration-job
    name: gcr.io/google.com/cloudsdktool/google-cloud-cli:slim
    entrypoint: bash
    args:
      - -ceu
      - |
        gcloud run jobs deploy "${_MIGRATION_JOB_NAME}" \
          --project="${PROJECT_ID}" \
          --region="${_REGION}" \
          --image="${_IMAGE_URI}" \
          --service-account="${_RUNWORKER_SERVICE_ACCOUNT}" \
          --set-cloudsql-instances="${_CLOUD_SQL_CONNECTION_NAME}" \
          --tasks=1 \
          --parallelism=1 \
          --max-retries=1 \
          --command=dagster \
          --args=instance,migrate \
          --set-env-vars="${_COMMON_ENV_VARS}" \
          --set-secrets="${_COMMON_SECRETS}" \
          --quiet

  # Dagster RunWorkerをデプロイ(パラメータはいい感じに)
  - id: deploy-run-worker
    name: gcr.io/google.com/cloudsdktool/google-cloud-cli:slim
    entrypoint: bash
    args:
      - -ceu
      # Cloud RunのJob名 を dagster.yaml の job_name_by_code_location と一致させること。
      - |
        gcloud run jobs deploy "dagster-run-worker" \
          --project="${PROJECT_ID}" \
          --region="${_REGION}" \
          --image="${_IMAGE_URI}" \
          --service-account="${_RUNWORKER_SERVICE_ACCOUNT}" \
          --set-cloudsql-instances="${_CLOUD_SQL_CONNECTION_NAME}" \
          --tasks=1 \
          --parallelism=1 \
          --max-retries=0 \
          --set-env-vars="${_COMMON_ENV_VARS}" \
          --set-secrets="${_COMMON_SECRETS}" \
          --quiet

  # Dagster Daemonをデプロイ(パラメータはいい感じに)
  - id: deploy-daemon-worker-pool
    name: gcr.io/google.com/cloudsdktool/google-cloud-cli:slim
    entrypoint: bash
    args:
      - -ceu
      - |
        gcloud run worker-pools deploy "dagster-daemon" \
          --project="${PROJECT_ID}" \
          --region="${_REGION}" \
          --image="${_IMAGE_URI}" \
          --service-account="${_DAEMON_SERVICE_ACCOUNT}" \
          --set-cloudsql-instances="${_CLOUD_SQL_CONNECTION_NAME}" \
          --command=dagster-daemon \
          --args=run,-w,/dagster/app/workspace.yaml \
          --set-env-vars="${_COMMON_ENV_VARS}" \
          --set-secrets="${_COMMON_SECRETS}" \
          --quiet

  # Dagster Webserverをデプロイ(Ingressや認証などパラメータはいい感じに)
  - id: deploy-webserver-plane
    name: gcr.io/google.com/cloudsdktool/google-cloud-cli:slim
    entrypoint: bash
    args:
      - -ceu
      - |
        gcloud run deploy "dagster-webserver" \
          --project="${PROJECT_ID}" \
          --region="${_REGION}" \
          --service-account="${_WEBSERVER_SERVICE_ACCOUNT}" \
          --set-cloudsql-instances="${_CLOUD_SQL_CONNECTION_NAME}" \
          --quiet \
          --container=webserver \
          --image="${_IMAGE_URI}" \
          --port=8080 \
          --command=dagster-webserver \
          --args=-h,0.0.0.0,-p,8080,-w,/dagster/app/workspace.yaml \
          --set-env-vars="${_COMMON_ENV_VARS}" \
          --set-secrets="${_COMMON_SECRETS}"

substitutions:
  _GOOGLE_CLOUD_PROJECT: ${PROJECT_ID}
  _REGION: asia-northeast1
  _RUNWORKER_SERVICE_ACCOUNT: "" # RunWorker用のサービスアカウント
  _DAEMON_SERVICE_ACCOUNT: "" # Daemon用のサービスアカウント
  _WEBSERVER_SERVICE_ACCOUNT: "" # Webserver用のサービスアカウント

  #.....

  _CLOUD_SQL_CONNECTION_NAME: "" # replace
  _POSTGRES_USER_SECRET: dagster-postgres-user
  _POSTGRES_DATABASE_SECRET: dagster-postgres-database
  _POSTGRES_PASSWORD_SECRET: dagster-postgres-password

  _BIGQUERY_LOCATION: ${_REGION}
  _GOOGLE_CLOUD_CREDENTIALS_SECRET: dagster-google-cloud-credentials

  _COMMON_ENV_VARS: DAGSTER_HOME=/dagster/dagster_home,DATA_PLATFORM_PROJECT_ROOT=/dagster/app,DATA_PLATFORM_ENV=prod,GOOGLE_CLOUD_PROJECT=${_GOOGLE_CLOUD_PROJECT},GOOGLE_CLOUD_REGION=${_REGION},PGHOST=/cloudsql/${_CLOUD_SQL_CONNECTION_NAME},BIGQUERY_LOCATION=${_BIGQUERY_LOCATION}
  _COMMON_SECRETS: DAGSTER_POSTGRES_USER=${_POSTGRES_USER_SECRET}:latest,DAGSTER_POSTGRES_DB=${_POSTGRES_DATABASE_SECRET}:latest,DAGSTER_POSTGRES_PASSWORD=${_POSTGRES_PASSWORD_SECRET}:latest,GOOGLE_CLOUD_CREDENTIALS=${_GOOGLE_CLOUD_CREDENTIALS_SECRET}:latest

options:
    dynamicSubstitutions: true

 ['${_IMAGE_URI}']

デプロイ

Cloud Buildを使ってデプロイ！

Bash
gcloud builds submit --config=cloudbuild.yaml .

Webserverから確認してみる

作成したCloud RunサービスのURLからDagsterの画面を見ることができました。

DagsterのJobsの画面

試しに実行してみても、動作しました。

成功しました！

DagsterのRunsの一覧画面

DagsterのRunsの詳細画面

終わりに

Cloud Runを使ったDagsterの構築をすることができました。

運用コストを抑えつつ利用することができて大満足です。

良いDagsterライフを

この記事を書いた人

Kanji Uema

Google CloudとPythonを活用し、データ基盤の設計・開発に取り組んでいます

プロフィールと記事一覧