Programar ejecuciones

En este documento, se muestra cómo hacer lo siguiente en Dataform:

En la siguiente tabla, se comparan los métodos:

Método Objetivo Tipo de activador
Parámetros de configuración del flujo de trabajo Son ejecuciones de producción estándar basadas en el tiempo dentro de Dataform. Relacionadas con el tiempo
Workflows y Cloud Scheduler Organización ligera y sin servidores de Dataform junto con otras APIs Basada en el tiempo o en eventos
Cloud Composer Canalizaciones complejas con dependencias fuera de BigQuery Basada en el tiempo o en eventos
Activadores de Cloud Build Automatizar las versiones inmediatamente después de un envío de Git Basada en eventos

Antes de comenzar

Para programar ejecuciones con configuraciones de flujo de trabajo o programar ejecuciones con flujos de trabajo y Cloud Scheduler, haz lo siguiente:

  1. En la consola de Google Cloud , ve a la página Dataform.

    Ir a Dataform

  2. Selecciona o crea un repositorio.

  3. Crea una configuración de lanzamiento.

Para programar ejecuciones con Cloud Composer, haz lo siguiente:

  1. Selecciona o crea un repositorio de Dataform.
  2. Otorga acceso de Dataform a BigQuery.
  3. Selecciona o crea un espacio de trabajo de Dataform.
  4. Crea al menos una tabla.
  5. Crea un entorno de Cloud Composer 2.

Roles obligatorios

Para obtener los permisos que necesitas para completar las tareas de este documento, pídele a tu administrador que te otorgue los siguientes roles de IAM:

Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

También puedes obtener los permisos necesarios a través de roles personalizados o cualquier otro rol predefinido.

Para usar una cuenta de servicio personalizada cuando crees una configuración de flujo de trabajo, otorga acceso a la cuenta de servicio personalizada.

Para usar las credenciales de usuario de la Cuenta de Google cuando crees una configuración de flujo de trabajo (vista previa), otorga acceso a la Cuenta de Google.

Para habilitar las ejecuciones programadas de una configuración de flujo de trabajo, debes otorgar el permiso iam.serviceAccounts.actAs al agente de servicio predeterminado de Dataform para la cuenta de servicio personalizada que se usa en la configuración del flujo de trabajo. Este permiso está disponible en el rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser). Para obtener más información, consulta Cómo usar el modo de suplantación estricto.

Para mejorar la seguridad de la programación, consulta Implementa permisos de programación mejorados.

Cómo programar ejecuciones con parámetros de configuración de flujos de trabajo

En esta sección, se muestra cómo crear una configuración de flujo de trabajo en Dataform para programar y configurar ejecuciones de flujos de trabajo. Puedes usar parámetros de configuración de flujo de trabajo para ejecutar flujos de trabajo de Dataform de manera programada.

Acerca de la configuración del flujo de trabajo

Para programar ejecuciones de Dataform de todas las acciones del flujo de trabajo o de las acciones seleccionadas en BigQuery, puedes crear parámetros de configuración del flujo de trabajo. En la configuración de un flujo de trabajo, seleccionas una configuración de versión de compilación, eliges las acciones del flujo de trabajo para la ejecución y estableces la programación de ejecución.

Luego, durante una ejecución programada de la configuración de tu flujo de trabajo, Dataform implementa tu selección de acciones del resultado de compilación más reciente en la configuración de tu versión en BigQuery. También puedes activar manualmente la ejecución de un parámetro de configuración de flujo de trabajo con workflowConfigs de la API de Dataform.

Una configuración del flujo de trabajo de Dataform contiene los siguientes parámetros de configuración de ejecución:

  • Es el ID de la configuración del flujo de trabajo.
  • Es la configuración de lanzamiento.

  • Cuenta de servicio.

    Esta es la cuenta de servicio personalizada asociada con la configuración del flujo de trabajo. Puedes seleccionar una cuenta de servicio personalizada asociada a tu proyecto Google Cloud o ingresar manualmente una cuenta de servicio diferente. De forma predeterminada, las configuraciones de flujos de trabajo usan las mismas cuentas de servicio que sus repositorios.

    Las credenciales de la cuenta de servicio son el método de autorización predeterminado para la creación y ejecución de configuraciones de flujos de trabajo programados.

  • Credenciales de usuario de la Cuenta de Google (vista previa)

    Las credenciales de usuario de la Cuenta de Google son el método de autorización predeterminado para la creación y ejecución manuales y no programadas de la configuración del flujo de trabajo. Para obtener más información, consulta Autoriza tu Cuenta de Google.

  • Acciones del flujo de trabajo que se ejecutarán:

    • Todas las acciones.
    • Selección de acciones
    • Selección de etiquetas

  • Esquema de ejecución y zona horaria.

Crea una configuración de flujo de trabajo

Para crear una configuración de flujo de trabajo de Dataform, sigue estos pasos:

  1. En tu repositorio, ve a Lanzamientos y programación.
  2. En la sección Configuración del flujo de trabajo, haz clic en Crear.

  3. En el panel Crear configuración de flujo de trabajo, en el campo ID de configuración, ingresa un ID único para la configuración del flujo de trabajo.

    Los IDs solo pueden incluir números, letras, guiones y guiones bajos.

  4. En el menú Configuración de lanzamiento, selecciona una configuración de lanzamiento de compilación.

  5. En la sección Autenticación, autoriza la configuración del flujo de trabajo con las credenciales de usuario de tu Cuenta de Google o una cuenta de servicio.

    • Para usar las credenciales de usuario de tu Cuenta de Google (versión preliminar), selecciona Ejecutar con mis credenciales de usuario.
    • Para usar una cuenta de servicio personalizada, selecciona Ejecutar con la cuenta de servicio seleccionada y, luego, selecciona la cuenta de servicio asociada con tu proyecto Google Cloud al que tienes acceso. Si no seleccionas una cuenta de servicio, la configuración del flujo de trabajo usará la cuenta de servicio del repositorio.
    Debes usar una cuenta de servicio personalizada o las credenciales de usuario de tu Cuenta de Google.

  6. Opcional: En el campo Frecuencia de programación, ingresa la frecuencia de las ejecuciones en el formato unix-cron.

    Para verificar que Dataform ejecute el resultado de compilación más reciente en la configuración de versión correspondiente, mantén un descanso de al menos una hora entre el momento de la creación del resultado de compilación y el momento de la ejecución programada.

  7. Opcional: En el menú Zona horaria, selecciona la zona horaria para las ejecuciones.

    La zona horaria predeterminada es UTC.

  8. Selecciona las acciones del flujo de trabajo que se ejecutarán:

    • Para ejecutar todo el flujo de trabajo, haz clic en Todas las acciones.
    • Para ejecutar las acciones seleccionadas en el flujo de trabajo, haz clic en Selección de acciones y, luego, selecciona las acciones.
    • Para ejecutar acciones con las etiquetas seleccionadas, haz clic en Selección de etiquetas y, luego, selecciona las etiquetas.
    • Opcional: Para ejecutar las acciones o etiquetas seleccionadas y sus dependencias, selecciona la opción Incluir dependencias.
    • Opcional: Para ejecutar las acciones o etiquetas seleccionadas y sus dependencias, selecciona la opción Incluir dependencias.

    • Opcional: Para volver a compilar todas las tablas desde cero, selecciona la opción Ejecutar con actualización completa.

      Sin esta opción, Dataform actualiza las tablas incrementales sin volver a compilarlas desde cero.

    • Opcional: Establece la prioridad del trabajo de consulta de BigQuery con la opción Ejecutar como trabajo interactivo con alta prioridad (predeterminada). De forma predeterminada, BigQuery ejecuta las consultas como trabajos de consulta interactivos, que están diseñados para comenzar a ejecutarse lo más rápido posible. Si borras esta opción, las consultas se ejecutarán como trabajos de consultas por lotes, que tienen una prioridad más baja.

  9. Haz clic en Crear. Si seleccionaste Ejecutar con mis credenciales de usuario como método de autenticación, debes autorizar tu Cuenta de Google (versión preliminar).

Por ejemplo, la siguiente configuración del flujo de trabajo ejecuta acciones con la etiqueta hourly cada hora en la zona horaria CEST:

  • ID de configuración: production-hourly
  • Configuración de lanzamiento: -
  • Frecuencia: 0 * * * *
  • Zona horaria: Central European Summer Time (CEST)
  • Selección de acciones del flujo de trabajo: Selección de etiquetas, etiqueta hourly

Autoriza tu Cuenta de Google

Para autenticar el recurso con las credenciales de usuario de tu Cuenta de Google, debes otorgar permiso de forma manual a las canalizaciones de BigQuery para obtener el token de acceso de tu Cuenta de Google y acceder a los datos de origen en tu nombre. Puedes otorgar la aprobación manual con la interfaz del diálogo de OAuth.

Solo debes otorgar permiso a las canalizaciones de BigQuery una vez.

Para revocar el permiso que otorgaste, sigue estos pasos:

  1. Ve a la página de tu Cuenta de Google.
  2. Haz clic en BigQuery Pipelines.
  3. Haz clic en Quitar acceso.

Cambiar el propietario de la configuración del flujo de trabajo actualizando las credenciales también requiere aprobación manual si el nuevo propietario de la Cuenta de Google nunca antes creó una configuración del flujo de trabajo.

Cómo editar la configuración de un flujo de trabajo

Para editar la configuración de un flujo de trabajo, sigue estos pasos:

  1. En tu repositorio, ve a Lanzamientos y programación.
  2. En la configuración del flujo de trabajo que deseas editar, haz clic en el menú Más y, luego, en Editar.
  3. En el panel Editar configuración del flujo de trabajo, edita la configuración del flujo de trabajo y, luego, haz clic en Guardar.

Borra la configuración de un flujo de trabajo

Para borrar la configuración de un flujo de trabajo, sigue estos pasos:

  1. En tu repositorio, ve a Lanzamientos y programación.
  2. En la configuración del flujo de trabajo que deseas borrar, haz clic en el menú Más y, luego, en Borrar.
  3. En el cuadro de diálogo Borrar configuración de lanzamiento, haz clic en Borrar.

Programa ejecuciones con Workflows y Cloud Scheduler

En esta sección, se muestra cómo programar ejecuciones de flujos de trabajo de Dataform con Workflows y Cloud Scheduler.

Acerca de las ejecuciones programadas del flujo de trabajo

Puedes establecer la frecuencia de las ejecuciones de tu flujo de trabajo de Dataform creando un trabajo de Cloud Scheduler que active un flujo de trabajo de Workflows. Workflows ejecuta servicios en un flujo de trabajo de organización que tú defines.

Workflows ejecuta tu flujo de trabajo de Dataform en un proceso de dos pasos. Primero, extrae el código de tu repositorio de Dataform de tu proveedor de Git y lo compila en un resultado de compilación. Luego, usa el resultado de la compilación para crear un flujo de trabajo de Dataform y lo ejecuta con la frecuencia que establezcas.

Crea un flujo de trabajo de orquestación programado

Para programar ejecuciones de tu flujo de trabajo de Dataform, usa Workflows para crear un flujo de trabajo de organización y agrega un trabajo de Cloud Scheduler como activador.

  1. Workflows usa cuentas de servicio para otorgar acceso a los flujos de trabajo a los recursos deGoogle Cloud . Crea una cuenta de servicio y otórgale los siguientes permisos:

  2. Crea un flujo de trabajo de organización y usa el siguiente código fuente en YAML como definición del flujo de trabajo:

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

    Reemplaza lo siguiente:

    • PROJECT_ID: Es el ID de tu proyecto de Google Cloud .
    • REPOSITORY_LOCATION: Es la ubicación de tu repositorio de Dataform.
    • REPOSITORY_ID: Es el nombre de tu repositorio de Dataform.
    • GIT_COMMITISH: Es la rama de Git desde la que deseas ejecutar el código de Dataform. Para un repositorio recién creado, reemplaza main.

  3. Programa el flujo de trabajo de organización con Cloud Scheduler.

Personaliza la solicitud de resultado de compilación de creación del flujo de trabajo de Dataform

Puedes actualizar el flujo de trabajo de orquestación existente y definir la configuración de la solicitud del resultado de la compilación de creación del flujo de trabajo de Dataform en formato YAML. Para obtener más información sobre la configuración, consulta la referencia del recurso projects.locations.repositories.compilationResults de la API de REST.

Por ejemplo, para agregar un parámetro de configuración de _dev schemaSuffix a todas las acciones durante la compilación, reemplaza el cuerpo del paso createCompilationResult por el siguiente fragmento de código:

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

También puedes pasar parámetros de configuración adicionales como argumentos de tiempo de ejecución en una solicitud de ejecución de Workflows y acceder a esos argumentos con variables. Para obtener más información, consulta Cómo pasar argumentos del entorno de ejecución en una solicitud de ejecución.

Personaliza la solicitud de invocación del flujo de trabajo de Dataform

Puedes actualizar el flujo de trabajo de orquestación existente y definir la configuración de la solicitud de invocación del flujo de trabajo de Dataform en formato YAML. Para obtener más información sobre la configuración de la solicitud de invocación, consulta la referencia del recurso de REST de projects.locations.repositories.workflowInvocations.

Por ejemplo, para ejecutar solo acciones con la etiqueta hourly con todas las dependencias transitivas incluidas, reemplaza el cuerpo de createWorkflowInvocation por el siguiente fragmento de código:

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

También puedes pasar parámetros de configuración adicionales como argumentos de tiempo de ejecución en una solicitud de ejecución de Workflows y acceder a esos argumentos con variables. Para obtener más información, consulta Cómo pasar argumentos del entorno de ejecución en una solicitud de ejecución.

Programa ejecuciones con Cloud Composer

Puedes usar Cloud Composer 2 para programar ejecuciones de Dataform. Dataform no admite Cloud Composer 1.

Para administrar las programaciones de las ejecuciones de Dataform con Cloud Composer 2, puedes usar operadores de Dataform en grafos acíclicos dirigidos (DAG) de Airflow. Puedes crear un DAG de Airflow que programe invocaciones de flujos de trabajo de Dataform.

Dataform proporciona varios operadores de Airflow. Estos incluyen operadores para obtener un resultado de compilación, obtener una invocación de flujo de trabajo y cancelar una invocación de flujo de trabajo. Para ver la lista completa de operadores de Airflow de Dataform disponibles, consulta Google Dataform Operators.

Instala el paquete google-cloud-dataform de PyPI

Si usas las versiones 2.0.25 y posteriores de Cloud Composer 2, este paquete ya está instalado en tu entorno. No es necesario que la instales.

Si usas versiones anteriores de Cloud Composer 2, instala el paquete google-cloud-dataform de PyPI.

En la sección Paquetes de PyPI, especifica la versión ==0.2.0.

Crea un DAG de Airflow que programe invocaciones de flujos de trabajo de Dataform

Para administrar las ejecuciones programadas de flujos de trabajo de Dataform con Cloud Composer 2, escribe el DAG con los operadores de Airflow de Dataform y, luego, súbelo al bucket de tu entorno.

En la siguiente muestra de código, se muestra un DAG de Airflow que crea un resultado de compilación de Dataform y, luego, inicia una invocación de flujo de trabajo de Dataform:

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

Reemplaza lo siguiente:

  • PROJECT_ID: Es el ID de tu proyecto de Dataform Google Cloud .
  • REPOSITORY_ID: Es el nombre de tu repositorio de Dataform.
  • REGION: Es la región en la que se encuentra el repositorio de Dataform.
  • COMPILATION_RESULT: Es el nombre del resultado de la compilación que deseas usar para esta invocación del flujo de trabajo.
  • GIT_COMMITISH: Es el commitish de Git en el repositorio de Git remoto de la versión del código que deseas usar, por ejemplo, una rama o un SHA de Git.

En el siguiente ejemplo de código, se muestra un DAG de Airflow que realiza las siguientes acciones:

  1. Crea un resultado de compilación de Dataform.
  2. Inicia una invocación asíncrona del flujo de trabajo de Dataform.
  3. Consulta el estado de tu flujo de trabajo hasta que ingrese al estado esperado con DataformWorkflowInvocationStateSensor.
from datetime import datetime

from google.cloud.dataform_v1beta1 import WorkflowInvocation

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

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,
    asynchronous=True,
    workflow_invocation={
        "compilation_result": COMPILATION_RESULT
    }
)

is_workflow_invocation_done = DataformWorkflowInvocationStateSensor(
    task_id="is_workflow_invocation_done",
    project_id=PROJECT_ID,
    region=REGION,
    repository_id=REPOSITORY_ID,
    workflow_invocation_id=("{{ task_instance.xcom_pull('create_workflow_invocation')['name'].split('/')[-1] }}"),
    expected_statuses={WorkflowInvocation.State.SUCCEEDED},
)


create_compilation_result >> create_workflow_invocation

Reemplaza lo siguiente:

  • PROJECT_ID: Es el ID de tu proyecto de Dataform Google Cloud .
  • REPOSITORY_ID: Es el nombre de tu repositorio de Dataform.
  • REGION: Es la región en la que se encuentra el repositorio de Dataform.
  • COMPILATION_RESULT: Es el nombre del resultado de la compilación que deseas usar para esta invocación del flujo de trabajo.
  • GIT_COMMITISH: Es el commitish de Git en el repositorio de Git remoto de la versión del código que deseas usar, por ejemplo, una rama o un SHA de Git.
  • COMPILATION_RESULT: Es el nombre del resultado de la compilación que deseas usar para esta invocación del flujo de trabajo.

Agrega parámetros de configuración de compilación

Puedes agregar parámetros de configuración de compilación adicionales al objeto create_compilation_result del DAG de Airflow. Para obtener más información sobre los parámetros disponibles, consulta la referencia de la API de CodeCompilationConfig de Dataform.

  • Para agregar parámetros de configuración de compilación al objeto create_compilation_result del DAG de Airflow, agrega los parámetros seleccionados al campo code_compilation_config con el siguiente formato:

        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,
            "code_compilation_config": { "PARAMETER": "PARAMETER_VALUE"}
        },
    )

    Reemplaza lo siguiente:

    • PROJECT_ID: Es el ID de tu proyecto de Dataform Google Cloud .
    • REPOSITORY_ID: Es el nombre de tu repositorio de Dataform.
    • REGION: Es la región en la que se encuentra el repositorio de Dataform.
    • GIT_COMMITISH: Es el commitish de Git en el repositorio de Git remoto de la versión del código que deseas usar, por ejemplo, una rama o un SHA de Git.
    • PARAMETER: Es el parámetroCodeCompilationConfig seleccionado. Puedes agregar varios parámetros.
    • PARAMETER_VALUE: Es el valor del parámetro seleccionado.

En el siguiente ejemplo de código, se muestra el parámetro defaultDatabase agregado al objeto create_compilation_result del DAG de Airflow:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": REMOTE_BRANCH,
            "code_compilation_config": { "default_database": "my-custom-gcp-project"}
        },
    )

Agrega parámetros de configuración de invocación del flujo de trabajo

Puedes agregar parámetros de configuración adicionales de invocación del flujo de trabajo al objeto DAG de create_workflow_invocation Airflow. Para obtener más información sobre los parámetros disponibles, consulta la referencia de la API de InvocationConfig de Dataform.

  • Para agregar parámetros de configuración de invocación del flujo de trabajo al objeto DAG de Airflow create_workflow_invocation, agrega los parámetros seleccionados al campo invocation_config con el siguiente formato:

        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'] }}",
            "invocation_config": { "PARAMETER": PARAMETER_VALUE }
        },
    )

    Reemplaza lo siguiente:

    • PROJECT_ID: Es el ID de tu proyecto de Dataform Google Cloud .
    • REPOSITORY_ID: Es el nombre de tu repositorio de Dataform.
    • REGION: Es la región en la que se encuentra el repositorio de Dataform.
    • PARAMETER: Es el parámetroInvocationConfig seleccionado. Puedes agregar varios parámetros.
    • PARAMETER_VALUE: Es el valor del parámetro seleccionado.

En el siguiente ejemplo de código, se muestran los parámetros includedTags[] y transitiveDependenciesIncluded agregados al objeto create_workflow_invocation del DAG de Airflow:

    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'] }}",
            "invocation_config": { "included_tags": ["daily"], "transitive_dependencies_included": true }
        },
    )

Automatiza las ejecuciones con activadores de Cloud Build

Si deseas ir más allá de las programaciones basadas en el tiempo en una configuración de versión, puedes usar un activador de Cloud Build para crear una canalización basada en eventos. Este enfoque compila automáticamente tu código cada vez que se envía una confirmación nueva a una rama de Git y, de inmediato, activa una invocación del flujo de trabajo de Dataform para actualizar tus datos.

Prepara tus recursos

  1. En tu proyecto Google Cloud , habilita las APIs de Dataform y Cloud Build:

    Habilita la API de Dataform

    Habilitar la API de Cloud Build

  2. Asegúrate de tener lo siguiente:

Otorga los permisos de IAM necesarios

Otorga el rol de administrador de Dataform (roles/dataform.admin) a la cuenta de servicio personalizada en tu repositorio de Dataform. Este rol proporciona acceso completo al repositorio, incluido el permiso para crear resultados de compilación, actualizar la configuración de lanzamiento y comenzar nuevas invocaciones de flujo de trabajo. Para obtener detalles sobre cómo otorgar un rol de IAM a un repositorio individual, consulta Controla el acceso a un repositorio individual.

Otorga el rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser) en la cuenta de servicio personalizada de la configuración del flujo de trabajo a la cuenta de servicio del activador de Cloud Build. Para obtener más información sobre este requisito, consulta Cómo usar el modo act-as estricto.

Para que Cloud Build use tu cuenta de servicio personalizada, debes otorgarle al agente de servicio de Cloud Build permiso para actuar como esa cuenta. Para otorgar permiso de suplantación al agente de servicio de Cloud Build, haz lo siguiente:

  1. En la consola de Google Cloud , ve a la página Cuentas de servicio.

    Ir a Cuentas de servicio

  2. Selecciona tu cuenta de servicio personalizada.

  3. Ve a la pestaña Principales con acceso.

  4. Haz clic en Otorgar acceso.

  5. En el campo Principales nuevos, ingresa la dirección de correo electrónico del agente de servicio de Cloud Build, que debe tener este formato:

    service-PROJECT_NUMBER@gcp-sa-cloudbuild.iam.gserviceaccount.com

    Reemplaza PROJECT_NUMBER por el ID numérico de tu proyecto deGoogle Cloud . Puedes encontrar el ID de tu proyecto Google Cloud en elGoogle Cloud panel de la consola. Para obtener más información, consulta Cómo encontrar el nombre, el número y el ID del proyecto.

  6. En el menú Selecciona un rol, selecciona Usuario de cuenta de servicio.

  7. Haz clic en Guardar.

Crea el archivo de configuración cloudbuild.yaml

En la raíz de tu repositorio de Git, crea un archivo cloudbuild.yaml. Usa este archivo para definir la siguiente secuencia de comandos de varios pasos para crear un resultado de compilación, actualizar la configuración de la versión para establecer este resultado de compilación como en vivo y comenzar una nueva invocación de flujo de trabajo.

steps:
  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk:latest'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e -o pipefail # Exit script on any error

        # 1. Get the access token
        TOKEN=$(gcloud auth print-access-token)

        # 2. Define API endpoints and resource names
        RELEASE_CONFIG_RESOURCE="projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/releaseConfigs/${_RELEASE_CONFIG_ID}"
        COMPILATION_RESULTS_API="https://dataform.googleapis.com/v1/projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/compilationResults"

        # 3. Create the new compilation result
        echo "Creating new compilation result from $$RELEASE_CONFIG_RESOURCE..."
        CREATE_PAYLOAD="{\"releaseConfig\": \"$$RELEASE_CONFIG_RESOURCE\"}"
        curl --fail-with-body -X POST \
          -H "Authorization: Bearer $$TOKEN" \
          -H "Content-Type: application/json" \
          -d "$$CREATE_PAYLOAD" \
          "$$COMPILATION_RESULTS_API" | tee /workspace/compilation_response.json

  - name: 'alpine'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e # Exit script on any error

        # 4. Parse compilation result name
        apk add --no-cache jq
        COMPILATION_NAME=$(jq -r '.name' < /workspace/compilation_response.json)

        echo "Successfully created compilation result: $$COMPILATION_NAME"
        echo $$COMPILATION_NAME > /workspace/compilation_result_name.txt

  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk:latest'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e # Exit script on any error
        # 5. Update the releaseConfig to set the new compilation result as 'live'
        COMPILATION_NAME=$(cat /workspace/compilation_result_name.txt)
        echo "Updating release config to set $$COMPILATION_NAME as live..."
        PATCH_PAYLOAD="{\"releaseCompilationResult\": \"$$COMPILATION_NAME\", \"gitCommitish\": \"$BRANCH_NAME\"}"

        RELEASE_CONFIG_RESOURCE="projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/releaseConfigs/${_RELEASE_CONFIG_ID}"
        RELEASE_CONFIG_PATCH_API="https://dataform.googleapis.com/v1/$${RELEASE_CONFIG_RESOURCE}"
        curl --fail-with-body -X PATCH \
          -H "Authorization: Bearer $(gcloud auth print-access-token)" \
          -H "Content-Type: application/json" \
          -d "$$PATCH_PAYLOAD" \
          "$$RELEASE_CONFIG_PATCH_API?updateMask=releaseCompilationResult"

        echo "Successfully updated release config."

  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk:latest'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        set -e # Exit script on any error
        # 6. Launch a workflow config after recompiling the release config
        WORKFLOW_CONFIG_RESOURCE="projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/workflowConfigs/${_WORKFLOW_CONFIG_ID}"
        CREATE_WORKFLOW_PAYLOAD="{\"workflowConfig\": \"$$WORKFLOW_CONFIG_RESOURCE\"}"

        WORKFLOW_INVOCATIONS_API="https://dataform.googleapis.com/v1/projects/${_PROJECT_ID}/locations/${_DATAFORM_LOCATION}/repositories/${_DATAFORM_REPO_ID}/workflowInvocations"
        curl --fail-with-body -X POST \
          -H "Authorization: Bearer $(gcloud auth print-access-token)" \
          -H "Content-Type: application/json" \
          -d "$$CREATE_WORKFLOW_PAYLOAD" \
          "$$WORKFLOW_INVOCATIONS_API"

        echo "Successfully created a new workflow invocation."

# Define substitution variables that can be set in the trigger
substitutions:
  _DATAFORM_LOCATION: 'us-central1'  # Default, change if needed
  _DATAFORM_REPO_ID: ''              # Required: Set this in the trigger
  _RELEASE_CONFIG_ID: ''             # Required: Set this in the trigger
  _WORKFLOW_CONFIG_ID: ''            # Required: Set this in the trigger
  _PROJECT_ID: ${PROJECT_ID}         # Automatically uses the build's Project ID

options:
  logging: CLOUD_LOGGING_ONLY

Crea el activador de Cloud Build

Para crear un activador que ejecute tu configuración de compilación cuando se envíe código a tu repositorio, haz lo siguiente:

  1. En la consola Google Cloud , abre la página Activadores de Cloud Build.

    Ir a Activadores

  2. Si no conectaste tu repositorio de Git, haz clic en Conectar repositorio y sigue los pasos.

  3. Haz clic en Crear activador.

  4. Ingresa un nombre para el activador.

  5. Selecciona una región para el activador.

  6. Selecciona un evento para el activador.

  7. En la sección Fuente, configura el repositorio en tu repositorio de Git conectado.

  8. Establece la rama principal de tu repositorio.

  9. En la sección Configuración, selecciona el archivo de configuración de Cloud Build, que puede ser un archivo YAML o JSON.

  10. Establece la ubicación del archivo en /cloudbuild.yaml o la ruta de acceso a tu archivo.

  11. En la sección Variables de sustitución, agrega las siguientes variables y valores:

    • _DATAFORM_REPO_ID: ID de tu repositorio de Dataform
    • _RELEASE_CONFIG_ID: ID de tu configuración de versión de Dataform
    • _WORKFLOW_CONFIG_ID: ID de configuración de tu flujo de trabajo de Dataform
    • Opcional: _DATAFORM_LOCATION: Es la región de tu repositorio de Dataform, por ejemplo, us-central1

  12. En la sección Cuenta de servicio, selecciona tu cuenta de servicio personalizada.

  13. Haz clic en Crear.

Para obtener más información, consulta Crea un activador de compilación.

Prueba el activador

  1. Confirma y envía el archivo cloudbuild.yaml a la rama que supervisa tu activador.

  2. Para ver la compilación de Cloud Build, abre la página Historial de compilación en la Google Cloud consola.

    Ir al historial de compilaciones

  3. Si la compilación se realiza correctamente, ve a la página Dataform.

    Ir a Dataform

  4. Selecciona tu repositorio.

  5. Haz clic en Lanzamientos y programación y selecciona la configuración de lanzamiento.

  6. En la lista Resultados de compilación manual / con API, busca una entrada nueva. La compilación correcta más reciente debe marcarse como el Resultado de compilación en vivo para la configuración de la versión.

  7. Haz clic en Registros de ejecución del flujo de trabajo.

  8. Deberías ver una nueva invocación del flujo de trabajo iniciada con la configuración de flujo de trabajo que seleccionaste.

¿Qué sigue?