Linhagem de dados com o Dataplex

Cloud Composer 1 | Cloud Composer 2

Sobre a integração da linhagem de dados

A linhagem de dados é um recurso do Dataplex que permite rastrear como os dados se movem pelos sistemas: de onde eles vêm, de onde são transmitidos e quais transformações são aplicadas a eles. A linhagem de dados está disponível para:

Depois que o recurso é ativado no ambiente do Cloud Composer, a execução de DAGs que utilizam qualquer um dos operadores compatíveis faz com que o Cloud Composer informe as informações de linhagem à API Data Lineage.

Você pode acessar essas informações usando:

Operadores compatíveis

Os operadores a seguir são compatíveis com relatórios de linhagem automáticos no Cloud Composer:

  • airflow.providers.google.cloud.operators.bigquery.BigQueryExecuteQueryOperator
  • airflow.providers.google.cloud.operators.bigquery.BigQueryInsertJobOperator
  • airflow.providers.google.cloud.transfers.bigquery_to_bigquery.BigQueryToBigQueryOperator
  • airflow.contrib.operators.bigquery_to_gcs.BigQueryToCloudStorageOperator
  • airflow.providers.google.cloud.transfers.bigquery_to_gcs.BigQueryToGCSOperator
  • airflow.providers.google.cloud.transfers.gcs_to_bigquery.GCSToBigQueryOperator
  • airflow.contrib.operators.gcs_to_bq.GoogleCloudStorageToBigQueryOperator
  • airflow.providers.google.cloud.operators.dataproc.DataprocSubmitJobOperator

Por exemplo, você executa a seguinte tarefa:

task = BigQueryInsertJobOperator(
    task_id='snapshot_task',
    dag=dag,
    location='<dataset-location>',
    configuration={
        'query': {
            'query': 'SELECT * FROM dataset.tableA',
            'useLegacySql': False,
            'destinationTable': {
                'project_id': GCP_PROJECT,
                'dataset_id': 'dataset',
                'table_id': 'tableB',
            },
        }
    },
)

Resulta na criação do seguinte gráfico de linhagem na IU do Dataplex:

Exemplo de gráfico de linhagem na interface do Dataplex.
Figura 1. Exemplo de gráfico de linhagem de uma tabela do BigQuery na interface do Dataplex.

Considerações sobre recursos do Cloud Composer

Cada execução de tarefa do Airflow que relata a linhagem de dados realiza:

  • Uma solicitação de RPC de criação ou atualização para um processo de linhagem
  • Uma solicitação de RPC de criação ou atualização para uma execução de linhagem
  • Uma ou mais solicitações de RPC para criar eventos de linhagem (na maioria das vezes 0 ou 1)

Para detalhes sobre essas entidades, consulte modelo de informações de linhagem e Referência da API Lineage na documentação do Dataplex.

O tráfego de linhagem emitido está sujeito a cotas na API Data Lineage. O Cloud Composer consome a cota de gravação.

O preço associado ao gerenciamento de dados de linhagem está sujeito ao preço de linhagem. Confira considerações sobre linhagem de dados.

Implicações no desempenho

A linhagem de dados é informada ao final da execução da tarefa do Airflow. Em média, o relatório da linhagem de dados leva de 1 a 2 segundos.

Isso não afeta o desempenho da tarefa: as tarefas do Airflow não falharão se a linhagem não for informada à API Lineage. Não há impacto na lógica do operador principal, mas toda a instância da tarefa é executada um pouco mais para considerar os dados de linhagem dos relatórios.

Um ambiente que informa a linhagem de dados terá um pequeno aumento nos custos associados, devido ao tempo extra necessário para relatar a linhagem de dados.

Compliance

A linhagem de dados oferece diferentes níveis de suporte para recursos como o VPC Service Controls. Analise as considerações sobre linhagem de dados para garantir que os níveis de suporte correspondam aos requisitos do ambiente.

Trabalhar com integração da linhagem de dados

A integração da linhagem de dados do Cloud Composer é gerenciada por ambiente. Isso significa que a ativação do recurso é feita em duas etapas:

  1. Ative a API Data Lineage no projeto.
  2. Ativar a integração da linhagem de dados em um ambiente específico do Cloud Composer.

Antes de começar

Ao criar um ambiente, a integração da linhagem de dados será ativada automaticamente se as seguintes condições forem atendidas:

  • A API Data Lineage está ativada no projeto. Para mais informações, consulte Como ativar a API Data Lineage na documentação do Dataplex.

  • Um back-end de linhagem personalizado não está configurado no Airflow.

  • A chave de criptografia gerenciada pelo cliente (CMEK) não está ativada no ambiente. A linhagem de dados não oferece suporte a CMEK para os metadados ingeridos. Não é possível ativar a integração da linhagem de dados nos ambientes do Cloud Composer que usam a CMEK. Para mais informações e outras limitações, consulte Considerações sobre linhagem de dados.

Para um ambiente atual, é possível enable ou desativar a integração da linhagem de dados a qualquer momento.

Funções exigidas

A integração com a linhagem de dados requer as seguintes permissões adicionadas à sua conta de serviço do ambiente do Cloud Composer:

  • Para as contas padrão de serviço: nenhuma alteração é necessária. As contas padrão de serviço incluem as permissões necessárias.
  • Para contas serviço gerenciado pelo usuário: conceda o papel de worker do Composer (roles/composer.worker) à conta de serviço. Esse papel inclui todas as permissões de linhagem de dados necessárias.

Para mais detalhes, consulte Permissões e papéis de linhagem na documentação do Dataplex.

Ativar a linhagem de dados no Cloud Composer

Console

  1. No console do Google Cloud, acesse a página Ambientes.

    Acessar "Ambientes"

  2. Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.

  3. Selecione a guia Configuração do ambiente.

  4. Na seção Integração da linhagem de dados do Dataplex, clique em Editar.

  5. No painel Integração da linhagem de dados do Dataplex, selecione Ativar integração com a linhagem de dados do Dataplex e clique em Salvar.

gcloud

Use o argumento --enable-cloud-data-lineage-integration.

gcloud composer environments update ENVIRONMENT_NAME \
    --location LOCATION \
    --enable-cloud-data-lineage-integration

Substitua:

  • ENVIRONMENT_NAME pelo nome do ambiente;

    O nome precisa começar com uma letra minúscula seguida por até 62 letras minúsculas, números ou hifens. Ele não pode terminar com um hífen. O nome do ambiente é usado para criar subcomponentes para o ambiente. Você precisa fornecer um nome que também seja válido como um nome de bucket do Cloud Storage. Para ver uma lista de restrições, consulte Diretrizes de nomenclatura de bucket.

  • LOCATION pela região do ambiente.

    Um local é a região em que o cluster do GKE do ambiente está localizado.

Exemplo:

gcloud composer environments update example-environment \
    --location us-central1 \
    --enable-cloud-data-lineage-integration

Enviar eventos de linhagem personalizados

É possível enviar eventos de linhagem personalizados se você quiser relatar a linhagem de um operador que não é compatível com relatórios de linhagem automatizados.

Por exemplo, para enviar eventos personalizados com:

  • BashOperator, modifique o parâmetro inlets ou outlets na definição da tarefa.
  • PythonOperator, modifique o parâmetro task.inlets ou task.outlets na definição da tarefa. Usar AUTO para o parâmetro inlets define o valor como o outlets da tarefa upstream.

Por exemplo, ao executar esta tarefa:


from airflow.composer.data_lineage.entities import BigQueryTable
from airflow.lineage import AUTO

…

bash_task = BashOperator(
   task_id='bash_task',
   dag=dag,
   bash_command='sleep 0',
   inlets=[BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table1',
   )],
   outlets=[BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table2',
   )]
)

def _python_task(task):
   task.inlets.append(BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table3',
   ))

   task.outlets.append(BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table4',
   ))

python_task = PythonOperator(
   task_id='python_task',
   dag=dag,
   python_callable=_python_task,
   inlets=[AUTO],
)

bash_task >> python_task

Resulta na criação do seguinte gráfico de linhagem na IU do Dataplex:

Exemplo de gráfico de linhagem para eventos personalizados na interface do Dataplex.
Figura 2. Gráfico de linhagem de amostra para várias tabelas do BigQuery na interface do Dataplex.

Desativar a linhagem de dados no Cloud Composer

Desativar a integração de linhagem em um ambiente do Cloud Composer não desativa a API Data Lineage. Se você quiser desativar completamente os relatórios de linhagem no projeto, desative também a API Data Lineage. Consulte Como desativar serviços.

Console

  1. No console do Google Cloud, acesse a página Ambientes.

    Acessar "Ambientes"

  2. Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.

  3. Selecione a guia Configuração do ambiente.

  4. Na seção Integração da linhagem de dados do Dataplex, clique em Editar.

  5. No painel Integração da linhagem de dados do Dataplex, selecione Desativar a integração com a linhagem de dados do Dataplex e clique em Salvar.

gcloud

Use o argumento --disable-cloud-data-lineage-integration.

gcloud composer environments update ENVIRONMENT_NAME \
    --location LOCATION \
    --disable-cloud-data-lineage-integration

Substitua:

  • ENVIRONMENT_NAME pelo nome do ambiente;

    O nome precisa começar com uma letra minúscula seguida por até 62 letras minúsculas, números ou hifens. Ele não pode terminar com um hífen. O nome do ambiente é usado para criar subcomponentes para o ambiente. Você precisa fornecer um nome que também seja válido como um nome de bucket do Cloud Storage. Para ver uma lista de restrições, consulte Diretrizes de nomenclatura de bucket.

  • LOCATION pela região do ambiente.

    Um local é a região em que o cluster do GKE do ambiente está localizado.

Exemplo:

gcloud composer environments update example-environment \
    --location us-central1 \
    --disable-cloud-data-lineage-integration

Conferir registros de linhagem no Cloud Composer

É possível inspecionar registros relacionados à linhagem de dados usando o link na página Configuração do ambiente na seção Integração da linhagem de dados do Dataplex.

Solução de problemas

Se os dados de linhagem não forem informados à API Lineage ou se você não conseguir vê-los no Dataplex, tente as seguintes etapas de solução de problemas:

  • Verifique se a API Data Lineage está ativada no projeto do ambiente do Cloud Composer.
  • Verifique se a integração da linhagem de dados está ativada no ambiente do Cloud Composer.
  • Verifique se o operador que você usa está incluído no suporte para geração automática de relatórios de linhagem. Consulte Operadores compatíveis com o Airflow.
  • Verifique se há possíveis problemas nos registros de linhagem no Cloud Composer.