デベロッパー

Workflows に並列ステップを導入: ステップの同時実行により、ワークフローの実行を高速化

2022年7月21日

https://storage.googleapis.com/gweb-cloudblog-publish/images/bigquery2022.max-2500x2500.jpg

Google Cloud Japan Team

※この投稿は米国時間 2022 年 7 月 12 日に、Google Cloud blog に投稿されたものの抄訳です。

このたび、Google は、新しい Workflows の機能をリリースいたします。これは、複数の Google Cloud と外部サービスを接続する、デベロッパー向けのサーバーレスオーケストレーターです。並列ステップは現在プレビュー版ですが、これを使用することで、デベロッパーが複数のステップを同時に実行することができ、特に HTTP リクエストやコールバックといった長時間実行オペレーションを含むワークフローの実行時間の短縮に役立ちます。

ワークフロー作成にあたり、デベロッパーは一連のステップと実行の順番を定義します。各ステップは変数を割り当てる、値を返す、HTTP エンドポイントを呼び出すなどのオペレーションを実行します。

デフォルトでは、Workflows は各ステップを 1 つずつ順番に実行します。このようにステップを 1 つずつ実行することは非効率と言えます。長時間実行オペレーションには、長時間実行する呼び出し、コールバック、ポーリング、人間の承認待ちなどが含まれ、数分、数時間、ときに数日かかることもあるからです。（Workflows の実行期間の上限は 1 年です。）

こうした非効率性を解決するため、Google は並列ブランチや並列反復処理を使用して各ステップを同時に実行する機能を追加しました。これにより、ワークフロー全体の実行時間を短縮できます。ワークフローに、順次オペレーション用の直列ステップと、非線形オペレーション用の並列ステップの両方を含めることができるようになったのです。

多くのユーザーにとって、並列ステップの Workflows は、Google Cloud でサービスのバッチを並列に実行し、結果を集計するのに最も効率的な方法です。Cloud Functions や Cloud Run などのサーバーレスコンピューティングサービスは自動スケーリングが可能なので、アイドル時に大容量をプロビジョニングすることなしに、必要なときに Workflows を使って高い同時実行性でこうしたサービスを実行できます。

この新機能について、さらに詳しく見ていきましょう。

並列ステップの実例: BigQuery ジョブの同時実行によるデータ処理の高速化

並列ステップのメリットを実証するために、Wikipedia のデータセット処理のための 5 つの BigQuery ジョブを実行するワークフローのチュートリアルの例をご紹介します。このチュートリアルでは、このワークフローの並列実行と非並列実行を比較し、これらの BigQuery ジョブを並列化した場合に実行時間が大幅に改善されることを確認します。

まず最初に、下記のループを使ってワークフローを順番に実行します。BigQuery の各ジョブは約 20 秒で実行され、トータルの実行時間は 1 分間ほどです。

直列反復処理

読み込んでいます...

steps:
    - init:
        assign:
            - results : {} # テーブル名をキーとした各反復処理の結果
            - tables:
                - 201201h
                - 201202h
                - 201203h
                - 201204h
                - 201205h
    - runQueries:
            for:
                value: table
                in: ${tables}
                steps:
                - runQuery:
                    call: googleapis.bigquery.v2.jobs.query
                    args:
                        projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                        body:
                            useLegacySql: false
                            useQueryCache: false
                            timeoutMs: 30000
                            # Wikipedia で閲覧数の多いタイトル上位 100 位を探す
                            query: ${
                                "SELECT TITLE, SUM(views)
                                FROM `bigquery-samples.wikipedia_pageviews." + table + "`
                                WHERE LENGTH(TITLE) > 10
                                GROUP BY TITLE
                                ORDER BY SUM(VIEWS) DESC
                                LIMIT 100"
                                }
                    result: queryResult
                - returnResult:
                    assign:
                        # 各テーブルの上位のタイトルを返す
                        - results[table]: {}
                        - results[table].title: ${queryResult.rows[0].f[0].v}
                        - results[table].views: ${queryResult.rows[0].f[1].v}
    - returnResults:
        return: ${results}

次に、BigQuery ジョブを同時に実行してみました。なお、非並列から並列反復処理への変更は非常に簡単に行えます。以下のように並列パラメータを追加し、結果を共有変数として宣言するだけです。

並列反復処理

読み込んでいます...

steps:
    - init:
        assign:
            - results : {} # テーブル名をキーとした各反復処理の結果
            - tables:
                - 201201h
                - 201202h
                - 201203h
                - 201204h
                - 201205h
    - runQueries:
        parallel:
            shared: [results]
            for:
                value: table
                in: ${tables}
                steps:
                - runQuery:
                    call: googleapis.bigquery.v2.jobs.query
                    args:
                        projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                        body:
                            useLegacySql: false
                            useQueryCache: false
                            timeoutMs: 30000
                            # Wikipedia で閲覧数の多いタイトル上位 100 位を探す
                            query: ${
                                "SELECT TITLE, SUM(views)
                                FROM `bigquery-samples.wikipedia_pageviews." + table + "`
                                WHERE LENGTH(TITLE) > 10
                                GROUP BY TITLE
                                ORDER BY SUM(VIEWS) DESC
                                LIMIT 100"
                                }
                    result: queryResult
                - returnResult:
                    assign:
                        # 各テーブルの上位のタイトルを返す
                        - results[table]: {}
                        - results[table].title: ${queryResult.rows[0].f[0].v}
                        - results[table].views: ${queryResult.rows[0].f[1].v}
    - returnResults:
        return: ${results}

並列反復処理で、これらの BigQuery ジョブを同時に実行するのにかかった時間は、合計 20 秒でした。これは非並列実行に比べると実に 5 倍の速さです。

BigQuery のジョブを並列実行することで、ワークフローの合計実行時間を 5 倍短縮できました。

この見積もりをツイートする

なお、並列ステップを使用する際には、すべての変数の割り当てはアトミックであることが保証されています。つまり、変数の読み書きの順序や競合状態を気にする必要はありません。上記の例のように、変数の共有を宣言することで、その変数が他のブランチにも書き込めるようになります。共有変数を使えば、他のブランチによる読み書きが介在することなく、割り当てられた値が決定され、書き込まれます。また、共有変数の書き込みは、他のブランチからもすぐに見られます。（ただし、実行順序は保証されていないことに注意してください。）

並列ブランチ: 一連のオペレーションを並列に実行する

ご自身のワークフローに、同時実行できる複数かつ異なるステップがある場合、並列ブランチに配置することで、それらのステップの完了に必要な合計時間を短縮できます。各並列ステップごとに最大 10 のブランチを定義でき、最大 20 のブランチを同時実行できます（20 を超えると、並列ブランチはキューに追加されます）。

以下で、並列ブランチを使用して、2 つの異なるサービスから並列にデータを取得する例を紹介します。

読み込んでいます...

main:
  params: [input]
  steps:
    - init:
        assign:
          - userProfile: {}
          - recentItems: []
    - enrichUserData:
        parallel:
          # userProfile と recentItems は、ブランチで書き込み可能にするために共有されています
          shared: [userProfile, recentItems]
          branches:
            - getUserProfileBranch:
                steps:
                  - getUserProfile:
                      call: http.get
                      args:
                        url: '${"https://example.com/users/" + input.userId}'
                      result: userProfile
            - getRecentItemsBranch:
                steps:
                  - getRecentItems:
                      try:
                        call: http.get
                        args:
                          url: '${"https://example.com/items?userId=" + input.userId}'
                        result: recentItems
                      except:
                        as: e
                        steps:
                          - ignoreError:
                              assign:
                                # この呼び出しが失敗した場合、空のリストで継続します
                                - recentItems: []

並列ステップを使用すべき理由

スリープ、HTTP リクエスト、コールバックなどの操作を含む、長時間（1 秒以上）実行するステップは、並列化により最も効率的になります。迅速に実行されるコンピューティングオペレーションや、assign、switch、next などの操作を含むステップについては、並列化しても効率が上がらないため、順番に実行する必要があります。

たとえば、先ほどのチュートリアルでは、BigQuery の各ジョブを実行するのに約 20 秒かかっています。これらのジョブは順番に実行する必要がないので、並列化により時間短縮することは非常に理にかなっています。

次のステップ: Workflows の並列ステップを使ってみる

並列ステップの Codelab をぜひご覧ください。この Codelab では、データセットを処理するために複数の BigQuery サービスを同時に実行する並列反復処理のチュートリアルを提供しています。
並列ステップやさらに多くのサンプルコードについて詳しくは、こちらのドキュメントをご覧ください。
Workflows を無料で試してみましょう。Workflows は従量課金制で、最初の内部ステップは 1 か月あたり 5,000 まで無料です。Google Cloud コンソールにアクセスすると利用開始できます。

並列ステップは現在プレビュー版です。この新機能を使用していてお気づきの点がありましたら、ぜひご意見をお聞かせください。こちらのフォームから、ご意見をお寄せください。

Google Cloud を利用している開発者やデータエンジニアの方で、まだ Workflows を使用していない方はぜひお試しください。イベントドリブンなビジネスプロセスやアプリケーション、または軽量なデータパイプラインを構築されたい場合は特におすすめします。Workflows について詳しくは、この無料の Codelab または Workflows プロダクトのドキュメントをご覧ください。

- Google Cloud、アウトバウンドプロダクトマネージャー Megan Bruce

投稿先