Ejecuta los pasos del flujo de trabajo en paralelo

Los pasos paralelos pueden reducir el tiempo de ejecución total de un flujo de trabajo, ya que realizan varias llamadas de bloqueo al mismo tiempo.

El bloqueo de llamadas, como sleep, llamadas HTTP y callbacks, puede demorar de milisegundos a días. El objetivo de los pasos paralelos es ayudar con esas operaciones simultáneas de larga duración. Si un flujo de trabajo debe realizar varias llamadas de bloqueo que sean independientes entre sí, el uso de ramas paralelas puede reducir el tiempo total de ejecución, ya que se inician las llamadas al mismo tiempo y se espera a que se completen todas.

Por ejemplo, si tu flujo de trabajo debe recuperar datos del cliente de varios sistemas independientes antes de continuar, las ramas paralelas permiten solicitudes simultáneas a la API. Si hay cinco sistemas y cada uno tarda dos segundos en responder, realizar los pasos de forma secuencial en un flujo de trabajo podría tardar al menos 10 segundos y realizarlos en paralelo podría tardar tan solo dos.

Nota: Las ramas paralelas no proporcionan ventajas de rendimiento para los pasos sin bloqueo, como asignaciones, condiciones o llamadas sin bloqueo. Debido a la sobrecarga del sistema, el uso de ramas paralelas que no incluyan llamadas de bloqueo puede aumentar el tiempo total de ejecución.

Crea un paso paralelo

Crea un paso parallel para definir una parte de tu flujo de trabajo en la que dos o más pasos se puedan ejecutar en simultáneo.

YAML

  - PARALLEL_STEP_NAME:
      parallel:
        exception_policy: POLICY
        shared: [VARIABLE_A, VARIABLE_B, ...]
        concurrency_limit: CONCURRENCY_LIMIT
        BRANCHES_OR_FOR:
          ...

JSON

  [
    {
      "PARALLEL_STEP_NAME": {
        "parallel": {
          "exception_policy": "POLICY",
          "shared": [
            "VARIABLE_A",
            "VARIABLE_B",
            ...
          ],
          "concurrency_limit": "CONCURRENCY_LIMIT",
          "BRANCHES_OR_FOR":
          ...
        }
      }
    }
  ]

Reemplaza lo siguiente:

• PARALLEL_STEP_NAME: Es el nombre del paso paralelo.
• POLICY (opcional): Determina la acción que realizarán otras ramas cuando se produzca una excepción no controlada. La política predeterminada, continueAll, no genera más acciones, y todas las otras ramas intentarán ejecutarse. Ten en cuenta que continueAll es la única política compatible por el momento.
• VARIABLE_A, VARIABLE_B, etc.: una lista de variables que admiten escritura con permiso superior que permiten asignaciones dentro del paso paralelo. Para obtener más información, consulta Variables compartidas.
• CONCURRENCY_LIMIT (opcional): Es la cantidad máxima de iteraciones y ramas que pueden ejecutarse de forma simultánea dentro de una sola ejecución de flujo de trabajo antes de que se agreguen más iteraciones y ramas a la cola para esperar. Esto se aplica a un único paso parallel y no se transmite en cascada. Debe ser un número entero positivo y puede ser un valor literal o una expresión. Para obtener más información, consulta Límites de simultaneidad.
• BRANCHES_OR_FOR: Usa branches o for para indicar una de las siguientes opciones:
  • Ramas que pueden ejecutarse en simultáneo.
  • Es un bucle en el que las iteraciones pueden ejecutarse en simultáneo.

Ten en cuenta lo siguiente:

• Las iteraciones y ramas paralelas pueden ejecutarse en cualquier orden y pueden hacerlo en un orden diferente con cada ejecución.
• Los pasos paralelos pueden incluir otros pasos paralelos anidados hasta el límite de profundidad. Consulta Cuotas y límites.
• Para obtener más información, consulta la página de referencia de sintaxis sobre pasos paralelos.

Sugerencia: Cloud Functions (2nd gen) puede aprovechar la simultaneidad a través del servicio subyacente de Cloud Run y darte más control sobre los precios. Cuando llamas a Cloud Functions en un paso for paralelo, puedes ahorrar costos si configuras la simultaneidad para que la misma instancia de Cloud Function procese varios elementos. Obtén más información para configurar la simultaneidad en Cloud Run.

Reemplaza la función experimental por un paso paralelo

Si usas experimental.executions.map para admitir el trabajo paralelo, puedes migrar el flujo de trabajo de modo que use pasos paralelos y ejecute bucles for comunes en paralelo. Para ver ejemplos, consulta Reemplaza la función experimental por un paso paralelo.

Muestras

Estos ejemplos demuestran la sintaxis.

Realiza operaciones en paralelo (con ramas)

Si tu flujo de trabajo tiene varios conjuntos de pasos diferentes que se pueden ejecutar al mismo tiempo, colocarlos en ramas paralelas puede disminuir el tiempo total necesario para completar esos pasos.

En el siguiente ejemplo, se pasa un ID de usuario como argumento al flujo de trabajo y los datos se recuperan en paralelo desde dos servicios diferentes. Las variables compartidas permiten que se escriban valores en las ramas y se los lea después de que se completen las ramas:

YAML

main:
  params: [input]
  steps:
    - init:
        assign:
          - userProfile: {}
          - recentItems: []
    - enrichUserData:
        parallel:
          shared: [userProfile, recentItems]  # userProfile and recentItems are shared to make them writable in the branches
          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:  # continue with an empty list if this call fails
                                - recentItems: []

JSON

{
  "main": {
    "params": [
      "input"
    ],
    "steps": [
      {
        "init": {
          "assign": [
            {
              "userProfile": {}
            },
            {
              "recentItems": []
            }
          ]
        }
      },
      {
        "enrichUserData": {
          "parallel": {
            "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": []
                                  }
                                ]
                              }
                            }
                          ]
                        }
                      }
                    }
                  ]
                }
              }
            ]
          }
        }
      }
    ]
  }
}

Procesa elementos en paralelo (con un bucle paralelo)

Si necesitas realizar la misma acción para cada elemento de una lista, puedes completar la ejecución más rápido con un bucle paralelo. Un bucle paralelo permite realizar varias iteraciones de bucle en paralelo. Ten en cuenta que, a diferencia de los bucles for regulares, las iteraciones se pueden realizar en cualquier orden.

En el siguiente ejemplo, se procesa un conjunto de notificaciones de usuario en un bucle for paralelo:

YAML

main:
  params: [input]
  steps:
    - sendNotifications:
        parallel:
          for:
            value: notification
            in: ${input.notifications}
            steps:
              - notify:
                  call: http.post
                  args:
                    url: https://example.com/sendNotification
                    body:
                      notification: ${notification}

JSON

{
  "main": {
    "params": [
      "input"
    ],
    "steps": [
      {
        "sendNotifications": {
          "parallel": {
            "for": {
              "value": "notification",
              "in": "${input.notifications}",
              "steps": [
                {
                  "notify": {
                    "call": "http.post",
                    "args": {
                      "url": "https://example.com/sendNotification",
                      "body": {
                        "notification": "${notification}"
                      }
                    }
                  }
                }
              ]
            }
          }
        }
      }
    ]
  }
}

Agregar datos (con un bucle paralelo)

Puedes procesar un conjunto de elementos mientras recopilas datos de las operaciones realizadas en cada elemento. Por ejemplo, es posible que desees hacer un seguimiento de los ID de los elementos creados o mantener una lista de elementos con errores.

En el siguiente ejemplo, 10 consultas distintas a un conjunto de datos públicos de BigQuery muestran la cantidad de palabras de un documento o conjunto de documentos. Una variable compartida permite que el recuento de palabras se acumule y se lea una vez que se completen todas las iteraciones. Después de calcular la cantidad de palabras en todos los documentos, el flujo de trabajo muestra el total.

YAML

main:
  params: [input]
  steps:
    - init:
        assign:
          - numWords: 0
          - corpuses:
              - sonnets
              - various
              - 1kinghenryvi
              - 2kinghenryvi
              - 3kinghenryvi
              - comedyoferrors
              - kingrichardiii
              - titusandronicus
              - tamingoftheshrew
              - loveslabourslost
    - runQueries:
        parallel:  # numWords is shared so it can be written within the parallel loop
          shared: [numWords]
          for:
            value: corpus
            in: ${corpuses}
            steps:
              - runQuery:
                  call: googleapis.bigquery.v2.jobs.query
                  args:
                    projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                    body:
                      useLegacySql: false
                      query: ${"SELECT COUNT(DISTINCT word) FROM `bigquery-public-data.samples.shakespeare` " + " WHERE corpus='" + corpus + "' "}
                  result: query
              - add:
                  assign:
                    - numWords: ${numWords + int(query.rows[0].f[0].v)}  # first result is the count
    - done:
        return: ${numWords}

JSON

{
  "main": {
    "params": [
      "input"
    ],
    "steps": [
      {
        "init": {
          "assign": [
            {
              "numWords": 0
            },
            {
              "corpuses": [
                "sonnets",
                "various",
                "1kinghenryvi",
                "2kinghenryvi",
                "3kinghenryvi",
                "comedyoferrors",
                "kingrichardiii",
                "titusandronicus",
                "tamingoftheshrew",
                "loveslabourslost"
              ]
            }
          ]
        }
      },
      {
        "runQueries": {
          "parallel": {
            "shared": [
              "numWords"
            ],
            "for": {
              "value": "corpus",
              "in": "${corpuses}",
              "steps": [
                {
                  "runQuery": {
                    "call": "googleapis.bigquery.v2.jobs.query",
                    "args": {
                      "projectId": "${sys.get_env(\"GOOGLE_CLOUD_PROJECT_ID\")}",
                      "body": {
                        "useLegacySql": false,
                        "query": "${\"SELECT COUNT(DISTINCT word) FROM `bigquery-public-data.samples.shakespeare` \" + \" WHERE corpus='\" + corpus + \"' \"}"
                      }
                    },
                    "result": "query"
                  }
                },
                {
                  "add": {
                    "assign": [
                      {
                        "numWords": "${numWords + int(query.rows[0].f[0].v)}"
                      }
                    ]
                  }
                }
              ]
            }
          }
        }
      },
      {
        "done": {
          "return": "${numWords}"
        }
      }
    ]
  }
}

¿Qué sigue?

• Referencia de la sintaxis: Pasos paralelos
• Instructivo: Ejecuta un flujo de trabajo que ejecute otros en paralelo
• Instructivo: Ejecuta múltiples trabajos de BigQuery en paralelo