Executar pipelines em um fluxo de trabalho

Você pode executar um pipeline como parte de um fluxo de trabalho de processamento de dados com Jobs do Lakeflow, Apache Airflow ou Azure Data Factory.

Jobs

Você pode orquestrar várias tarefas em um trabalho do Databricks para implementar um fluxo de trabalho de processamento de dados. Para incluir um pipeline em uma tarefa, utilize a tarefa Pipeline ao criar uma tarefa. Consulte Tarefa de pipeline para trabalhos.

Fluxo de ar do Apache

O Apache Airflow é uma solução de software livre para gerenciar e agendar fluxos de trabalho de dados. Airflow representa fluxos de trabalho como grafos acíclicos direcionados (DAGs) de operações. Você define um fluxo de trabalho em um arquivo Python e o Airflow gerencia o agendamento e a execução. Para obter informações sobre como instalar e usar o Airflow com o Azure Databricks, consulte Orchestrate Lakeflow Jobs com o Apache Airflow.

Para executar um pipeline como parte de um fluxo de trabalho do Airflow, use DatabricksSubmitRunOperator.

Requirements

Você precisará do seguinte para usar o suporte do Airflow a Pipelines Declarativos do Lakeflow Spark:

• Airflow versão 2.1.0 ou posterior.
• O pacote do provedor do Databricks versão 2.1.0 ou posterior.

Example

O exemplo a seguir cria um DAG de Airflow que aciona uma atualização para o pipeline com o identificador 8279d543-063c-4d63-9926-dae38e35ce8b:

from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator
from airflow.utils.dates import days_ago

default_args = {
  'owner': 'airflow'
}

with DAG('ldp',
         start_date=days_ago(2),
         schedule_interval="@once",
         default_args=default_args
         ) as dag:

  opr_run_now=DatabricksSubmitRunOperator(
    task_id='run_now',
    databricks_conn_id='CONNECTION_ID',
    pipeline_task={"pipeline_id": "8279d543-063c-4d63-9926-dae38e35ce8b"}
  )

Substitua CONNECTION_ID pelo identificador de uma Conexão do Airflow com seu workspace.

Salve este exemplo no diretório airflow/dags e use a UI do Airflow para visualizar e acionar o DAG. Utilize a interface do usuário da linha de produção para ver os detalhes da atualização da linha de produção.

Azure Data Factory

O Azure Data Factory é um serviço ETL baseado em nuvem que permite orquestrar fluxos de trabalho de integração e transformação de dados. O Azure Data Factory dá suporte diretamente à execução de tarefas do Azure Databricks em um fluxo de trabalho, incluindo notebooks, tarefas JAR e scripts Python. Você também pode incluir um pipeline em um fluxo de trabalho chamando a API REST do pipeline de uma atividade da Web do Azure Data Factory. Por exemplo, para disparar uma atualização de pipeline do Azure Data Factory:

1. Crie um data factory ou abra um data factory existente.

2. Quando a criação for concluída, abra a página da sua fábrica de dados e clique no botão Abrir o Azure Data Factory Studio. A interface do usuário do Azure Data Factory é exibida.

3. Crie um pipeline do Azure Data Factory selecionando Pipeline no menu suspenso Novo na interface do usuário do Azure Data Factory Studio.

4. Na caixa de ferramentas Atividades, expanda Geral e arraste a atividade da Web para a tela do pipeline. Clique na guia Configurações e insira os seguintes valores:

   Observação

   Como prática recomendada de segurança, quando você se autentica com ferramentas automatizadas, sistemas, scripts e aplicativos, o Databricks recomenda que você use tokens de acesso pessoal pertencentes a entidades de serviço em vez de usuários do workspace. Para criar tokens para entidades de serviço, consulte Gerenciar tokens para uma entidade de serviço.

   • URL: https://<databricks-instance>/api/2.0/pipelines/<pipeline-id>/updates.

     Substitua <get-workspace-instance>.

     Substitua <pipeline-id> pelo identificador de pipeline.

   • Método: selecione POST no menu suspenso.

   • Cabeçalhos: Clique + Novo. Na caixa de texto Nome, insira Authorization. Na caixa de texto Valor , insira Bearer <personal-access-token>.

     Substitua <personal-access-token> por um token de acesso pessoal do Azure Databricks.

   • Corpo: para passar parâmetros de solicitação adicionais, insira um documento JSON contendo os parâmetros. Por exemplo, para iniciar uma atualização e reprocessar todos os dados do pipeline: {"full_refresh": "true"}. Se não houver parâmetros de solicitação adicionais, insira chaves vazias ({}).

Para testar a atividade da Web, clique em Depurar na barra de ferramentas do pipeline na interface do usuário do Data Factory. A saída e o status da execução, incluindo erros, são exibidos na guia Saída do pipeline do Azure Data Factory. Use a interface do usuário dos pipelines para exibir os detalhes da atualização do pipeline.