Escreva DAGs do Airflow

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Este guia mostra como escrever um gráfico acíclico dirigido (DAG) do Apache Airflow que é executado num ambiente do Cloud Composer.

Uma vez que o Apache Airflow não oferece um forte isolamento de DAGs e tarefas, recomendamos que use ambientes de produção e de teste separados para evitar interferências de DAGs. Para mais informações, consulte o artigo Testar DAGs.

Estruturar um DAG do Airflow

Um DAG do Airflow é definido num ficheiro Python e é composto pelos seguintes componentes:

  • Definição do DAG
  • Operadores do Airflow
  • Relações com operadores

Os seguintes fragmentos de código mostram exemplos de cada componente fora do contexto.

Uma definição de DAG

O exemplo seguinte demonstra uma definição de DAG do Airflow:

Airflow 2 

import datetime

from airflow import models

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# 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(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

Fluxo de ar 1 

import datetime

from airflow import models

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# 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(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

Operadores e tarefas

Os operadores do Airflow descrevem o trabalho a realizar. Uma tarefa task é uma instância específica de um operador.

Airflow 2 

from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = PythonOperator(task_id="hello", python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = BashOperator(task_id="bye", bash_command="echo Goodbye.")

Fluxo de ar 1 

from airflow.operators import bash_operator
from airflow.operators import python_operator

    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

Relações de tarefas

As relações entre tarefas descrevem a ordem pela qual o trabalho tem de ser concluído.

Airflow 2 

# Define the order in which the tasks complete by using the >> and <<
# operators. In this example, hello_python executes before goodbye_bash.
hello_python >> goodbye_bash

Fluxo de ar 1 

# Define the order in which the tasks complete by using the >> and <<
# operators. In this example, hello_python executes before goodbye_bash.
hello_python >> goodbye_bash

Exemplo de fluxo de trabalho DAG completo em Python

O fluxo de trabalho seguinte é um modelo DAG funcional completo composto por duas tarefas: uma tarefa hello_python e uma tarefa goodbye_bash:

Airflow 2 


import datetime

from airflow import models

from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator



default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# 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(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = PythonOperator(task_id="hello", python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = BashOperator(task_id="bye", bash_command="echo Goodbye.")

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Fluxo de ar 1 


import datetime

from airflow import models

from airflow.operators import bash_operator
from airflow.operators import python_operator



default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    "start_date": datetime.datetime(2018, 1, 1),
}

# 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(
    "composer_sample_simple_greeting",
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    def greeting():
        import logging

        logging.info("Hello World!")

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id="hello", python_callable=greeting
    )

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id="bye", bash_command="echo Goodbye."
    )

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Para mais informações sobre a definição de DAGs do Airflow, consulte o tutorial do Airflow e os conceitos do Airflow.

Operadores do Airflow

Os exemplos seguintes mostram alguns operadores populares do Airflow. Para uma referência autorizada dos operadores do Airflow, consulte a referência de operadores e hooks e o índice de fornecedores.

BashOperator

Use o BashOperator para executar programas de linha de comandos.

Airflow 2 

from airflow.operators import bash

    # Create BigQuery output dataset.
    make_bq_dataset = bash.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f"bq ls {bq_dataset_name} || bq mk {bq_dataset_name}",
    )

Fluxo de ar 1 

from airflow.operators import bash_operator

    # Create BigQuery output dataset.
    make_bq_dataset = bash_operator.BashOperator(
        task_id="make_bq_dataset",
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f"bq ls {bq_dataset_name} || bq mk {bq_dataset_name}",
    )

O Cloud Composer executa os comandos fornecidos num script Bash num worker do Airflow. O trabalhador é um contentor Docker baseado no Debian e inclui vários pacotes.

PythonOperator

Use o PythonOperator para executar código Python arbitrário.

O Cloud Composer executa o código Python num contentor que inclui pacotes para a versão da imagem do Cloud Composer usada no seu ambiente.

Para instalar pacotes Python adicionais, consulte o artigo Instalar dependências do Python.

Google Cloud Operadores

Para executar tarefas que usam produtos Google Cloud , use os operadores doGoogle Cloud Airflow. Por exemplo, os operadores do BigQuery consultam e processam dados no BigQuery.

Existem muitos mais operadores do Airflow para o Google Cloud e serviços individuais fornecidos pela Google Cloud. Consulte a secção Google Cloud Operadores para ver a lista completa.

Airflow 2 

from airflow.providers.google.cloud.operators import bigquery
from airflow.providers.google.cloud.transfers import bigquery_to_gcs

    bq_recent_questions_query = bigquery.BigQueryInsertJobOperator(
        task_id="bq_recent_questions_query",
        configuration={
            "query": {
                "query": RECENT_QUESTIONS_QUERY,
                "useLegacySql": False,
                "destinationTable": {
                    "projectId": project_id,
                    "datasetId": bq_dataset_name,
                    "tableId": bq_recent_questions_table_id,
                },
            }
        },
        location=location,
    )

Fluxo de ar 1 

from airflow.contrib.operators import bigquery_operator

    # Query recent StackOverflow questions.
    bq_recent_questions_query = bigquery_operator.BigQueryOperator(
        task_id="bq_recent_questions_query",
        sql="""
        SELECT owner_display_name, title, view_count
        FROM `bigquery-public-data.stackoverflow.posts_questions`
        WHERE creation_date < CAST('{max_date}' AS TIMESTAMP)
            AND creation_date >= CAST('{min_date}' AS TIMESTAMP)
        ORDER BY view_count DESC
        LIMIT 100
        """.format(
            max_date=max_query_date, min_date=min_query_date
        ),
        use_legacy_sql=False,
        destination_dataset_table=bq_recent_questions_table_id,
    )

EmailOperator

Use o EmailOperator para enviar emails a partir de um DAG. Para enviar emails a partir de um ambiente do Cloud Composer, configure o seu ambiente para usar o SendGrid.

Airflow 2 

from airflow.operators import email

    # Send email confirmation (you will need to set up the email operator
    # See https://cloud.google.com/composer/docs/how-to/managing/creating#notification
    # for more info on configuring the email operator in Cloud Composer)
    email_summary = email.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

Fluxo de ar 1 

from airflow.operators import email_operator

    # Send email confirmation
    email_summary = email_operator.EmailOperator(
        task_id="email_summary",
        to="{{var.value.email}}",
        subject="Sample BigQuery notify data ready",
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][0] }}"
            ),
            view_count=(
                "{{ ti.xcom_pull(task_ids='bq_read_most_popular', "
                "key='return_value')[0][1] }}"
            ),
            export_location=output_file,
        ),
    )

Notificações sobre falhas do operador

Defina email_on_failure como True para enviar uma notificação por email quando um operador no DAG falhar. Para enviar notificações por email a partir de um ambiente do Cloud Composer, tem de configurar o seu ambiente para usar o SendGrid.

Airflow 2 

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": project_id,
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

Fluxo de ar 1 

from airflow import models

default_dag_args = {
    "start_date": yesterday,
    # Email whenever an Operator in the DAG fails.
    "email": "{{var.value.email}}",
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": "{{var.value.gcp_project}}",
}

with models.DAG(
    "composer_sample_bq_notify",
    schedule_interval=datetime.timedelta(weeks=4),
    default_args=default_dag_args,
) as dag:

Diretrizes do fluxo de trabalho de DAG

  • Coloque quaisquer bibliotecas Python personalizadas num arquivo ZIP de um DAG num diretório aninhado. Não coloque bibliotecas no nível superior do diretório DAGs.

    Quando o Airflow analisa a pasta dags/, verifica apenas os DAGs nos módulos Python que se encontram no nível superior da pasta DAGs e no nível superior de um arquivo ZIP também localizado na pasta dags/ de nível superior. Se o Airflow encontrar um módulo Python num arquivo ZIP que não contenha as subcadeias airflow e DAG, o Airflow para de processar o arquivo ZIP. O Airflow devolve apenas os DAGs encontrados até esse ponto.

  • Use o Airflow 2 em vez do Airflow 1.

    A comunidade do Airflow já não publica novas versões secundárias ou de patches para o Airflow 1.

  • Para tolerância a falhas, não defina vários objetos DAG no mesmo módulo Python.

  • Não use SubDAGs. Em alternativa, agrupe as tarefas em DAGs.

  • Coloque os ficheiros necessários no momento da análise DAG na pasta dags/ e não na pasta data/.

  • Implemente testes de unidades para os seus DAGs.

  • Teste os DAGs desenvolvidos ou modificados, conforme recomendado nas instruções para testar DAGs.

  • Verifique se os DAGs desenvolvidos não aumentam demasiado os tempos de análise dos DAGs.

  • As tarefas do Airflow podem falhar por vários motivos. Para evitar falhas de execuções de DAGs completas, recomendamos que ative as novas tentativas de tarefas. Se definir o número máximo de tentativas como 0, não são feitas tentativas.

    Recomendamos que substitua a opção default_task_retries por um valor para as novas tentativas de tarefas diferente de 0. Além disso, pode definir o parâmetro retries ao nível da tarefa.

  • Se quiser usar a GPU nas suas tarefas do Airflow, crie um cluster do GKE separado com base em nós que usem máquinas com GPUs. Use o GKEStartPodOperator para executar as suas tarefas.

  • Evite executar tarefas com utilização intensiva da CPU e da memória no conjunto de nós do cluster onde estão a ser executados outros componentes do Airflow (programadores, trabalhadores, servidores Web). Em alternativa, use KubernetesPodOperator ou GKEStartPodOperator.

  • Quando implementar DAGs num ambiente, carregue apenas os ficheiros que são absolutamente necessários para interpretar e executar DAGs na pasta /dags.

  • Limite o número de ficheiros DAG na pasta /dags.

    O Airflow está a analisar continuamente os DAGs na pasta /dags. A análise é um processo que percorre a pasta DAGs e o número de ficheiros que têm de ser carregados (com as respetivas dependências) afeta o desempenho da análise DAG e do agendamento de tarefas. É muito mais eficiente usar 100 ficheiros com 100 DAGs cada do que 10 000 ficheiros com 1 DAG cada. Por isso, recomenda-se esta otimização. Esta otimização é um equilíbrio entre o tempo de análise e a eficiência da criação e gestão de DAGs.

    Também pode considerar, por exemplo, implementar 10 000 ficheiros DAG. Para tal, pode criar 100 ficheiros ZIP, cada um com 100 ficheiros DAG.

    Além das sugestões acima, se tiver mais de 10 000 ficheiros DAG, a geração de DAGs de forma programática pode ser uma boa opção. Por exemplo, pode implementar um único ficheiro DAG Python que gera um determinado número de objetos DAG (por exemplo, 20 ou 100 objetos DAG).

  • Evite usar operadores do Airflow descontinuados. Em alternativa, use as respetivas alternativas atualizadas.

Perguntas frequentes sobre a escrita de DAGs

Como posso minimizar a repetição de código se quiser executar tarefas iguais ou semelhantes em vários DAGs?

Sugerimos que defina bibliotecas e wrappers para minimizar a repetição de código.

Como posso reutilizar código entre ficheiros DAG?

Coloque as funções de utilidade numa biblioteca Python local e importe as funções. Pode fazer referência às funções em qualquer DAG localizado na pasta dags/ no contentor do seu ambiente.

Como posso minimizar o risco de surgirem definições diferentes?

Por exemplo, tem duas equipas que querem agregar dados não processados em métricas de receita. As equipas escrevem duas tarefas ligeiramente diferentes que fazem a mesma coisa. Defina bibliotecas para trabalhar com os dados de receita, de modo que os implementadores do DAG tenham de esclarecer a definição da receita que está a ser agregada.

Como posso definir dependências entre DAGs?

Isto depende da forma como quer definir a dependência.

Se tiver dois DAGs (DAG A e DAG B) e quiser que o DAG B seja acionado após o DAG A, pode colocar um TriggerDagRunOperator no final do DAG A.

Se o DAG B depender apenas de um artefacto gerado pelo DAG A, como uma mensagem do Pub/Sub, um sensor pode funcionar melhor.

Se o DAG B estiver estreitamente integrado com o DAG A, pode conseguir unir os dois DAGs num único DAG.

Como posso transmitir IDs de execução exclusivos a um DAG e às respetivas tarefas?

Por exemplo, quer transmitir nomes de clusters do Dataproc e caminhos de ficheiros.

Pode gerar um ID exclusivo aleatório devolvendo str(uuid.uuid4()) num PythonOperator. Isto coloca o ID em XComs para que possa consultar o ID noutros operadores através de campos baseados em modelos.

Antes de gerar um uuid, pondere se um ID específico do DagRun seria mais valioso. Também pode fazer referência a estes IDs em substituições Jinja usando macros.

Como posso separar tarefas num DAG?

Cada tarefa deve ser uma unidade de trabalho idempotente. Consequentemente, deve evitar encapsular um fluxo de trabalho de vários passos numa única tarefa, como um programa complexo em execução num PythonOperator.

Devo definir várias tarefas num único DAG para agregar dados de várias origens?

Por exemplo, tem várias tabelas com dados não processados e quer criar agregados diários para cada tabela. As tarefas não dependem umas das outras. Deve criar uma tarefa e um DAG para cada tabela ou criar um DAG geral?

Se não se importar que cada tarefa partilhe as mesmas propriedades ao nível do DAG, como schedule_interval, faz sentido definir várias tarefas num único DAG. Caso contrário, para minimizar a repetição de código, podem ser gerados vários DAGs a partir de um único módulo Python, colocando-os no globals() do módulo.

Como posso limitar o número de tarefas simultâneas em execução num DAG?

Por exemplo, quer evitar exceder os limites/quotas de utilização da API ou evitar a execução de demasiados processos simultâneos.

Pode definir pools do Airflow na IU Web do Airflow e associar tarefas a pools existentes nos seus DAGs.

Perguntas frequentes sobre a utilização de operadores

Devo usar o DockerOperator?

Não recomendamos a utilização do comando DockerOperator, a menos que seja usado para iniciar contentores numa instalação remota do Docker (não dentro do cluster de um ambiente). Num ambiente do Cloud Composer, o operador não tem acesso a daemons do Docker.

Em alternativa, use KubernetesPodOperator ou GKEStartPodOperator. Estes operadores iniciam pods do Kubernetes em clusters do Kubernetes ou do GKE, respetivamente. Tenha em atenção que não recomendamos o lançamento de pods no cluster de um ambiente, porque isto pode levar a concorrência de recursos.

Devo usar o SubDagOperator?

Não recomendamos a utilização do SubDagOperator.

Use alternativas, conforme sugerido em Agrupar tarefas.

Devo executar código Python apenas em PythonOperators para separar totalmente os operadores Python?

Consoante o seu objetivo, tem algumas opções.

Se a sua única preocupação for manter dependências Python separadas, pode usar o comando PythonVirtualenvOperator.

Considere usar a KubernetesPodOperator. Este operador permite-lhe definir pods do Kubernetes e executar os pods noutros clusters.

Como posso adicionar pacotes binários personalizados ou não pertencentes ao PyPI?

Pode instalar pacotes alojados em repositórios de pacotes privados.

Como posso transmitir argumentos de forma uniforme a um DAG e às respetivas tarefas?

Pode usar o suporte integrado do Airflow para a criação de modelos Jinja para transmitir argumentos que podem ser usados em campos com modelos.

Quando ocorre a substituição de modelos?

A substituição de modelos ocorre nos trabalhadores do Airflow imediatamente antes de a função pre_execute de um operador ser chamada. Na prática, isto significa que os modelos não são substituídos até pouco antes da execução de uma tarefa.

Como posso saber que argumentos de operador suportam a substituição de modelos?

Os argumentos dos operadores que suportam a substituição de modelos Jinja2 estão explicitamente marcados como tal.

Procure o campo template_fields na definição do operador, que contém uma lista de nomes de argumentos que são substituídos por modelos.

Por exemplo, consulte o elemento BashOperator, que suporta a utilização de modelos para os argumentos bash_command e env.

Operadores do Airflow descontinuados e removidos

Os operadores do Airflow indicados na tabela seguinte estão descontinuados:

  • Evite usar estes operadores nos seus DAGs. Em alternativa, use os operadores de substituição atualizados fornecidos.

  • Se um operador estiver listado como disponível, significa que a versão de manutenção mais recente do Cloud Composer (1.20.12) ainda tem este operador disponível.

  • Alguns dos operadores de substituição não são suportados em nenhuma versão do Cloud Composer 1. Para os usar, considere fazer a atualização para o Cloud Composer 3 ou o Cloud Composer 2.

Operador descontinuado Estado Operador de substituição Substituição disponível a partir de
CreateAutoMLTextTrainingJobOperator Disponível na versão 1.20.12 SupervisedFineTuningTrainOperator O operador de substituição não está disponível
GKEDeploymentHook Disponível na versão 1.20.12 GKEKubernetesHook O operador de substituição não está disponível
GKECustomResourceHook Disponível na versão 1.20.12 GKEKubernetesHook O operador de substituição não está disponível
GKEPodHook Disponível na versão 1.20.12 GKEKubernetesHook O operador de substituição não está disponível
GKEJobHook Disponível na versão 1.20.12 GKEKubernetesHook O operador de substituição não está disponível
GKEPodAsyncHook Disponível na versão 1.20.12 GKEKubernetesAsyncHook O operador de substituição não está disponível
SecretsManagerHook Disponível na versão 1.20.12 GoogleCloudSecretManagerHook O operador de substituição não está disponível
BigQueryExecuteQueryOperator Disponível na versão 1.20.12 BigQueryInsertJobOperator Disponível na versão 1.20.12
BigQueryPatchDatasetOperator Disponível na versão 1.20.12 BigQueryUpdateDatasetOperator Disponível na versão 1.20.12
DataflowCreateJavaJobOperator Disponível na versão 1.20.12 beam.BeamRunJavaPipelineOperator Disponível na versão 1.20.12
DataflowCreatePythonJobOperator Disponível na versão 1.20.12 beam.BeamRunPythonPipelineOperator Disponível na versão 1.20.12
DataprocSubmitPigJobOperator Disponível na versão 1.20.12 DataprocSubmitJobOperator Disponível na versão 1.20.12
DataprocSubmitHiveJobOperator Disponível na versão 1.20.12 DataprocSubmitJobOperator Disponível na versão 1.20.12
DataprocSubmitSparkSqlJobOperator Disponível na versão 1.20.12 DataprocSubmitJobOperator Disponível na versão 1.20.12
DataprocSubmitSparkJobOperator Disponível na versão 1.20.12 DataprocSubmitJobOperator Disponível na versão 1.20.12
DataprocSubmitHadoopJobOperator Disponível na versão 1.20.12 DataprocSubmitJobOperator Disponível na versão 1.20.12
DataprocSubmitPySparkJobOperator Disponível na versão 1.20.12 DataprocSubmitJobOperator Disponível na versão 1.20.12
BigQueryTableExistenceAsyncSensor Disponível na versão 1.20.12 BigQueryTableExistenceSensor O operador de substituição não está disponível
BigQueryTableExistencePartitionAsyncSensor Disponível na versão 1.20.12 BigQueryTablePartitionExistenceSensor O operador de substituição não está disponível
CloudComposerEnvironmentSensor Disponível na versão 1.20.12 CloudComposerCreateEnvironmentOperator, CloudComposerDeleteEnvironmentOperator, CloudComposerUpdateEnvironmentOperator O operador de substituição não está disponível
GCSObjectExistenceAsyncSensor Disponível na versão 1.20.12 GCSObjectExistenceSensor O operador de substituição não está disponível
GoogleAnalyticsHook Disponível na versão 1.20.12 GoogleAnalyticsAdminHook O operador de substituição não está disponível
GoogleAnalyticsListAccountsOperator Disponível na versão 1.20.12 GoogleAnalyticsAdminListAccountsOperator O operador de substituição não está disponível
GoogleAnalyticsGetAdsLinkOperator Disponível na versão 1.20.12 GoogleAnalyticsAdminGetGoogleAdsLinkOperator O operador de substituição não está disponível
GoogleAnalyticsRetrieveAdsLinksListOperator Disponível na versão 1.20.12 GoogleAnalyticsAdminListGoogleAdsLinksOperator O operador de substituição não está disponível
GoogleAnalyticsDataImportUploadOperator Disponível na versão 1.20.12 GoogleAnalyticsAdminCreateDataStreamOperator O operador de substituição não está disponível
GoogleAnalyticsDeletePreviousDataUploadsOperator Disponível na versão 1.20.12 GoogleAnalyticsAdminDeleteDataStreamOperator O operador de substituição não está disponível
DataPipelineHook Disponível na versão 1.20.12 DataflowHook O operador de substituição não está disponível
CreateDataPipelineOperator Disponível na versão 1.20.12 DataflowCreatePipelineOperator O operador de substituição não está disponível
RunDataPipelineOperator Disponível na versão 1.20.12 DataflowRunPipelineOperator O operador de substituição não está disponível
AutoMLDatasetLink Disponível na versão 1.20.12 TranslationLegacyDatasetLink O operador de substituição não está disponível
AutoMLDatasetListLink Disponível na versão 1.20.12 TranslationDatasetListLink O operador de substituição não está disponível
AutoMLModelLink Disponível na versão 1.20.12 TranslationLegacyModelLink O operador de substituição não está disponível
AutoMLModelTrainLink Disponível na versão 1.20.12 TranslationLegacyModelTrainLink O operador de substituição não está disponível
AutoMLModelPredictLink Disponível na versão 1.20.12 TranslationLegacyModelPredictLink O operador de substituição não está disponível
AutoMLBatchPredictOperator Disponível na versão 1.20.12 vertex_ai.batch_prediction_job O operador de substituição não está disponível
AutoMLPredictOperator Disponível na versão 1.20.12 vertex_aigenerative_model. TextGenerationModelPredictOperator, translate.TranslateTextOperator O operador de substituição não está disponível
PromptLanguageModelOperator Disponível na versão 1.20.12 TextGenerationModelPredictOperator O operador de substituição não está disponível
GenerateTextEmbeddingsOperator Disponível na versão 1.20.12 TextEmbeddingModelGetEmbeddingsOperator O operador de substituição não está disponível
PromptMultimodalModelOperator Disponível na versão 1.20.12 GenerativeModelGenerateContentOperator O operador de substituição não está disponível
PromptMultimodalModelWithMediaOperator Disponível na versão 1.20.12 GenerativeModelGenerateContentOperator O operador de substituição não está disponível
DataflowStartSqlJobOperator Disponível na versão 1.20.12 DataflowStartYamlJobOperator O operador de substituição não está disponível
LifeSciencesHook Disponível na versão 1.20.12 Hook dos operadores do Google Cloud Batch A anunciar
DataprocScaleClusterOperator Disponível na versão 1.20.12 DataprocUpdateClusterOperator A anunciar
MLEngineStartBatchPredictionJobOperator Disponível na versão 1.20.12 CreateBatchPredictionJobOperator A anunciar
MLEngineManageModelOperator Disponível na versão 1.20.12 MLEngineCreateModelOperator, MLEngineGetModelOperator A anunciar
MLEngineGetModelOperator Disponível na versão 1.20.12 GetModelOperator A anunciar
MLEngineDeleteModelOperator Disponível na versão 1.20.12 DeleteModelOperator A anunciar
MLEngineManageVersionOperator Disponível na versão 1.20.12 MLEngineCreateVersion, MLEngineSetDefaultVersion, MLEngineListVersions, MLEngineDeleteVersion A anunciar
MLEngineCreateVersionOperator Disponível na versão 1.20.12 Parâmetro parent_model para operadores do VertexAI A anunciar
MLEngineSetDefaultVersionOperator Disponível na versão 1.20.12 SetDefaultVersionOnModelOperator A anunciar
MLEngineListVersionsOperator Disponível na versão 1.20.12 ListModelVersionsOperator A anunciar
MLEngineDeleteVersionOperator Disponível na versão 1.20.12 DeleteModelVersionOperator A anunciar
MLEngineStartTrainingJobOperator Disponível na versão 1.20.12 CreateCustomPythonPackageTrainingJobOperator A anunciar
MLEngineTrainingCancelJobOperator Disponível na versão 1.20.12 CancelCustomTrainingJobOperator A anunciar
LifeSciencesRunPipelineOperator Disponível na versão 1.20.12 Operadores do Google Cloud Batch A anunciar
MLEngineCreateModelOperator Disponível na versão 1.20.12 operador do VertexAI correspondente A anunciar

O que se segue?