Configurar a ingestão de linhagem de dados para um serviço

É possível ativar ou desativar seletivamente a ingestão de linhagem para integrações específicas no nível do projeto, da pasta ou da organização.

Para mais informações sobre integrações e cenários de configuração compatíveis, consulte Controlar a ingestão de linhagem de dados.

Pré-requisitos

Para controlar a ingestão de linhagem, use a API Data Lineage. Verifique se você tem um projeto cliente configurado para faturamento e cota, porque a API Data Lineage é uma API baseada em cliente.

Papéis e permissões

Para receber as permissões necessárias para configurar e controlar a ingestão de linhagem de dados, peça ao administrador para conceder a você os seguintes papéis do Identity and Access Management (IAM):

Para receber as permissões necessárias para configurar e controlar a ingestão de linhagem de dados, peça ao administrador para conceder a você os seguintes papéis do IAM no projeto:

Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esses papéis predefinidos contêm as permissões necessárias para configurar e controlar a ingestão de linhagem de dados. Para acessar as permissões exatas necessárias, expanda a seção Permissões necessárias:

Permissões necessárias

As seguintes permissões são necessárias para configurar e controlar a ingestão de linhagem de dados:

  • Receber e atualizar configurações de linhagem:
    • datalineage.configs.get
    • datalineage.configs.update

Essas permissões também podem ser concedidas com funções personalizadas ou outros papéis predefinidos.

  1. Ative a API datalineage.googleapis.com no projeto do cliente. Para mais informações, consulte Ativar a linhagem de dados.

  2. Defina o projeto do cliente. Para os exemplos a seguir, use o cabeçalho X-Goog-User-Project. Para mais informações, consulte Parâmetros do sistema.

Receber configuração atual

Para verificar se a ingestão de linhagem está ativada para um recurso ou para receber o valor etag antes de modificar a configuração, recupere a configuração atual.

C#

C#

Antes de testar esta amostra, siga as instruções de configuração do C# no Guia de início rápido do Knowledge Catalog: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Knowledge Catalog C#.

Para autenticar no Knowledge Catalog, configure o Application Default Credentials. Para mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local. 

using Google.Cloud.DataCatalog.Lineage.ConfigManagement.V1;

public sealed partial class GeneratedConfigManagementServiceClientSnippets
{
    /// <summary>Snippet for GetConfig</summary>
    /// <remarks>
    /// This snippet has been automatically generated and should be regarded as a code template only.
    /// It will require modifications to work:
    /// - It may require correct/in-range values for request initialization.
    /// - It may require specifying regional endpoints when creating the service client as shown in
    ///   https://cloud.google.com/dotnet/docs/reference/help/client-configuration#endpoint.
    /// </remarks>
    public void GetConfigRequestObject()
    {
        // Create client
        ConfigManagementServiceClient configManagementServiceClient = ConfigManagementServiceClient.Create();
        // Initialize request argument(s)
        GetConfigRequest request = new GetConfigRequest
        {
            ConfigName = ConfigName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
        };
        // Make the request
        Config response = configManagementServiceClient.GetConfig(request);
    }
}

Go

Go

Antes de testar esta amostra, siga as instruções de configuração do Go no Guia de início rápido do Knowledge Catalog: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Knowledge Catalog Go.

Para autenticar no Knowledge Catalog, configure o Application Default Credentials. Para mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local. 


//go:build examples

package main

import (
	"context"

	configmanagement "cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1"
	configmanagementpb "cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1/configmanagementpb"
)

func main() {
	ctx := context.Background()
	// This snippet has been automatically generated and should be regarded as a code template only.
	// It will require modifications to work:
	// - It may require correct/in-range values for request initialization.
	// - It may require specifying regional endpoints when creating the service client as shown in:
	//   https://pkg.go.dev/cloud.google.com/go#hdr-Client_Options
	c, err := configmanagement.NewClient(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	defer c.Close()

	req := &configmanagementpb.GetConfigRequest{
		// TODO: Fill request struct fields.
		// See https://pkg.go.dev/cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1/configmanagementpb#GetConfigRequest.
	}
	resp, err := c.GetConfig(ctx, req)
	if err != nil {
		// TODO: Handle error.
	}
	// TODO: Use resp.
	_ = resp
}

Java

Java

Antes de testar esta amostra, siga as instruções de configuração do Java no Guia de início rápido do Knowledge Catalog: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Knowledge Catalog Java.

Para autenticar no Knowledge Catalog, configure o Application Default Credentials. Para mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local. 

import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;

public class SyncGetConfig {

  public static void main(String[] args) throws Exception {
    syncGetConfig();
  }

  public static void syncGetConfig() throws Exception {
    // This snippet has been automatically generated and should be regarded as a code template only.
    // It will require modifications to work:
    // - It may require correct/in-range values for request initialization.
    // - It may require specifying regional endpoints when creating the service client as shown in
    // https://cloud.google.com/java/docs/setup#configure_endpoints_for_the_client_library
    try (ConfigManagementServiceClient configManagementServiceClient =
        ConfigManagementServiceClient.create()) {
      GetConfigRequest request =
          GetConfigRequest.newBuilder()
              .setName(ConfigName.ofProjectLocationName("[PROJECT]", "[LOCATION]").toString())
              .build();
      Config response = configManagementServiceClient.getConfig(request);
    }
  }
}

Python

Python

Antes de testar esta amostra, siga as instruções de configuração do Python no Guia de início rápido do Knowledge Catalog: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Knowledge Catalog Python.

Para autenticar no Knowledge Catalog, configure o Application Default Credentials. Para mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local. 

# This snippet has been automatically generated and should be regarded as a
# code template only.
# It will require modifications to work:
# - It may require correct/in-range values for request initialization.
# - It may require specifying regional endpoints when creating the service
#   client as shown in:
#   https://googleapis.dev/python/google-api-core/latest/client_options.html
from google.cloud import datacatalog_lineage_configmanagement_v1


def sample_get_config():
    # Create a client
    client = datacatalog_lineage_configmanagement_v1.ConfigManagementServiceClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_configmanagement_v1.GetConfigRequest(
        name="name_value",
    )

    # Make the request
    response = client.get_config(request=request)

    # Handle the response
    print(response)

gcloud

Para conferir a configuração de linhagem atual, use o comando gcloud datalineage config describe. É possível recuperar a configuração de um projeto, uma pasta ou uma organização.

O exemplo a seguir mostra como receber a configuração do projeto atual:

gcloud datalineage config describe

Por exemplo, para receber a configuração de um projeto específico, use a flag --project:

gcloud datalineage config describe --project=PROJECT_ID

Substitua:

  • PROJECT_ID: o ID do projeto cuja configuração você quer ver.

Para conferir a configuração atual de ingestão de linhagem de um serviço em uma pasta ou organização, substitua --project=PROJECT_ID por um dos seguintes:

  • --folder=FOLDER_ID se você quiser ver as configurações de ingestão de dados de uma pasta.
  • --organization=ORGANIZATION_ID se você quiser ver as configurações de ingestão de dados de uma organização.

REST

Para conferir a configuração de linhagem de dados atual, use o método projects.locations.config.get. É possível recuperar a configuração de um projeto, uma pasta ou uma organização.

O exemplo a seguir mostra como receber a configuração de um projeto:

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

  • CLIENT_PROJECT_ID: o ID do projeto do cliente usado para faturamento ou cotas.
  • PROJECT_ID: o ID do projeto cuja configuração você quer ver.

Método HTTP e URL: 

GET https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

Para enviar a solicitação, expanda uma destas opções:

O comando retorna uma das seguintes saídas:

  • Se você não fornecer nenhuma configuração de ingestão de linhagem, vai receber uma saída com um objeto ingestion vazio: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {}
}

    Isso significa que o serviço usa a configuração padrão de ingestão de linhagem. Neste exemplo, a configuração de ingestão de linhagem do Serviço Gerenciado para Apache Spark é enabled.

  • Se você ativar a ingestão de linhagem explicitamente, vai receber a seguinte saída: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}
  • Se a ingestão de linhagem estiver desativada, você vai receber a seguinte saída: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Para receber a configuração de uma pasta ou organização, substitua projects/"PROJECT_ID por folders/FOLDER_ID ou organizations/ORGANIZATION_ID.

O campo etag na resposta é um checksum gerado pelo servidor com base no valor atual da configuração. Ao atualizar uma configuração usando o método patch, você pode incluir o valor etag retornado de uma solicitação get recente no corpo da solicitação. Se você fornecer o etag, o Knowledge Catalog vai usá-lo para verificar se a configuração não mudou desde sua última solicitação de leitura. Se houver uma incompatibilidade, a solicitação de atualização vai falhar. Isso evita que você substitua sem querer as configurações feitas por outros usuários em cenários de leitura-modificação-gravação. Se você não fornecer um etag na solicitação patch, o Knowledge Catalog vai substituir a configuração incondicionalmente.

Desativar a ingestão de linhagem para um serviço

Para gerenciar custos, aplicar políticas de governança de dados ou excluir projetos de desenvolvimento e outras cargas de trabalho que não se beneficiam do rastreamento de linhagem, desative a ingestão de linhagem para um serviço.

Java

package com.google.cloud.datacatalog.lineage.configmanagement.v1.samples;

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.LineageEnablement;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.UpdateConfigRequest;

public class DisableLineageIngestion {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String location = "global";
    disableLineageIngestion(projectId, location);
  }

  // Disables lineage ingestion for a specific service
  // (Managed Service for Apache Spark).
  public static void disableLineageIngestion(String projectId, String location) throws Exception {
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests.
    try (ConfigManagementServiceClient client = ConfigManagementServiceClient.create()) {
      // Format the resource name.
      String name = ConfigName.ofProjectLocationName(projectId, location).toString();

      Config.Builder configBuilder = Config.newBuilder().setName(name);

      // It is a best practice to read the existing config to preserve other rules
      // and use the etag for optimistic concurrency control.
      try {
        GetConfigRequest getRequest = GetConfigRequest.newBuilder().setName(name).build();
        Config existingConfig = client.getConfig(getRequest);
        configBuilder.mergeFrom(existingConfig);
      } catch (NotFoundException e) {
        // If config doesn't exist, we will proceed by creating a new one.
      }

      // Create an integration selector for the service you want to disable.
      IntegrationSelector selector =
          IntegrationSelector.newBuilder().setIntegration(Integration.DATAPROC).build();

      // Set lineage enablement to false to disable tracking.
      LineageEnablement enablement = LineageEnablement.newBuilder().setEnabled(false).build();

      // Build the ingestion rule.
      IngestionRule disableRule =
          IngestionRule.newBuilder()
              .setIntegrationSelector(selector)
              .setLineageEnablement(enablement)
              .build();

      // Preserve existing rules except for the one we are modifying, then add the new rule.
      // We clear the ingestion block out of the configBuilder entirely to reconstruct it.
      Ingestion.Builder ingestionBuilder = Ingestion.newBuilder();
      if (configBuilder.hasIngestion()) {
        for (IngestionRule rule : configBuilder.getIngestion().getRulesList()) {
          // Keep all existing rules EXCEPT the one targeting DATAPROC
          if (rule.getIntegrationSelector().getIntegration() != Integration.DATAPROC) {
            ingestionBuilder.addRules(rule);
          }
        }
      }
      ingestionBuilder.addRules(disableRule);

      // Update the config builder with the reconstructed ingestion settings.
      configBuilder.setIngestion(ingestionBuilder.build());

      // Build the update request.
      UpdateConfigRequest request = UpdateConfigRequest.newBuilder()
          .setConfig(configBuilder.build())
          .build();

      // Update the config.
      Config response = client.updateConfig(request);
      System.out.printf("Successfully updated config: %s\n", response.getName());
    }
  }
}

Python

from google.api_core.exceptions import NotFound
from google.cloud.datacatalog.lineage import configmanagement_v1

def disable_lineage_ingestion(project_id: str, location: str = "global") -> configmanagement_v1.Config:
    """Disables lineage ingestion for a specific service.

    Args:
        project_id: The ID of your Google Cloud project.
        location: The region location, usually 'global'.

    Returns:
        The updated Configuration object.
    """
    # Initialize client that will be used to send requests.
    client = configmanagement_v1.ConfigManagementServiceClient()

    # The config name format
    name = f"projects/{project_id}/locations/{location}/config"

    try:
        # Retrieve the existing config to preserve other configurations and
        # obtain the latest etag for optimistic concurrency control.
        config = client.get_config(name=name)

        # Filter out existing rules for the integration we are updating
        new_rules = [
            rule for rule in config.ingestion.rules
            if rule.integration_selector.integration != configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
        ]
    except NotFound:
        # If the config does not exist, start fresh
        config = configmanagement_v1.Config(name=name)
        new_rules = []

    # Define the integration to disable tracking for (e.g., DATAPROC).
    integration_selector = configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector(
        integration=configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
    )

    # Set lineage enablement to False to disable tracking.
    lineage_enablement = configmanagement_v1.Config.Ingestion.IngestionRule.LineageEnablement(
        enabled=False
    )

    # Create the ingestion rule.
    disable_rule = configmanagement_v1.Config.Ingestion.IngestionRule(
        integration_selector=integration_selector,
        lineage_enablement=lineage_enablement,
    )
     # Append the new disabling rule and assign it back to the config ingestion rules
    new_rules.append(disable_rule)
    config.ingestion = configmanagement_v1.Config.Ingestion(rules=new_rules)

    # Create the update request using the config (which includes the etag if it existed).
    request = configmanagement_v1.UpdateConfigRequest(
        config=config,
    )

    # Make the request to update the config
    response = client.update_config(request=request)

    print(f"Successfully updated config: {response.name}")
    return response

gcloud

Para desativar a ingestão de linhagem de dados em um serviço específico, use o comando gcloud datalineage config update com uma string JSON inline ou um caminho para um arquivo JSON que define lineageEnablement.enabled como false para o integration específico.

O exemplo a seguir mostra como desativar a ingestão de linhagem de um serviço para um projeto usando uma string JSON inline:

gcloud datalineage config update --project=PROJECT_ID \
  --config='{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "INTEGRATION"
          },
          "lineageEnablement": {
            "enabled": false
          }
        }
      ]
    },
    "etag": "ETAG"
  }'

Substitua:

  • PROJECT_ID: o ID do projeto cuja configuração você quer atualizar.
  • INTEGRATION: a integração para a qual você definiu a configuração. Por exemplo, DATAPROC ou BIGQUERY.
  • ETAG: o valor etag retornado de uma solicitação get recente no corpo da solicitação, usado para verificar se a configuração não mudou desde sua última solicitação de leitura.

Para atualizar a configuração usando um arquivo JSON, execute:

gcloud datalineage config update --project=PROJECT_ID --config=CONFIG_FILE

Substitua:

  • CONFIG_FILE: o caminho para o arquivo JSON que contém a configuração.

Para desativar a ingestão de linhagem de um serviço em uma pasta ou organização, substitua --project=PROJECT_ID por uma das seguintes opções:

  • --folder=FOLDER_ID se quiser atualizar as configurações de ingestão de dados de uma pasta.
  • --organization=ORGANIZATION_ID se você quiser atualizar as configurações de ingestão de dados de uma organização.

REST

Para desativar a ingestão de linhagem de dados em um serviço específico, use o método projects.locations.config.patch com uma regra de ingestão que defina lineageEnablement.enabled como false para o integration específico.

Para evitar a substituição acidental de configurações feitas por outros usuários em cenários de leitura-modificação-gravação, inclua o campo etag no corpo da solicitação. Para mais informações, consulte Receber a configuração atual.

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

  • CLIENT_PROJECT_ID: o ID do projeto do cliente usado para faturamento ou cotas.
  • PROJECT_ID: o ID do projeto cuja configuração você quer atualizar.
  • ETAG: o valor etag retornado de uma solicitação get recente.
  • INTEGRATION: o integration para o qual você definiu a configuração. Por exemplo, DATAPROC.

Método HTTP e URL: 

PATCH https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

Corpo JSON da solicitação:

{
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "ETAG"
}

Para enviar a solicitação, expanda uma destas opções:

Você receberá uma resposta JSON semelhante a esta:

{
  "name": "projects/PROJECT_ID/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Para desativar a ingestão de linhagem em uma pasta ou organização, substitua projects/"PROJECT_ID por folders/FOLDER_ID ou organizations/ORGANIZATION_ID.

Ativar a ingestão de linhagem para um serviço

Para retomar o rastreamento depois de desativá-lo ou ativar uma integração que está desativada por padrão, ative a ingestão de linhagem para um serviço.

Java

package com.google.cloud.datacatalog.lineage.configmanagement.v1.samples;

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.LineageEnablement;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.UpdateConfigRequest;

public class EnableLineageIngestion {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String location = "global";
    enableLineageIngestion(projectId, location);
  }

  // Enables lineage ingestion for a specific service
  // (Managed Service for Apache Spark).
  public static void enableLineageIngestion(String projectId, String location) throws Exception {
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests.
    try (ConfigManagementServiceClient client = ConfigManagementServiceClient.create()) {
      // Format the resource name.
      String name = ConfigName.ofProjectLocationName(projectId, location).toString();

      Config.Builder configBuilder = Config.newBuilder().setName(name);

      // It is a best practice to read the existing config to preserve other rules
      // and use the etag for optimistic concurrency control.
      try {
        GetConfigRequest getRequest = GetConfigRequest.newBuilder().setName(name).build();
        Config existingConfig = client.getConfig(getRequest);
        configBuilder.mergeFrom(existingConfig);
      } catch (NotFoundException e) {
        // If config doesn't exist, we will proceed by creating a new one.
      }

      // Create an integration selector for the service you want to enable (e.g., DATAPROC).
      IntegrationSelector selector =
          IntegrationSelector.newBuilder().setIntegration(Integration.DATAPROC).build();

      // Set lineage enablement to true to enable tracking.
      LineageEnablement enablement = LineageEnablement.newBuilder().setEnabled(true).build();

      // Build the ingestion rule.
      IngestionRule enableRule =
          IngestionRule.newBuilder()
              .setIntegrationSelector(selector)
              .setLineageEnablement(enablement)
              .build();

      // Preserve existing rules except for the one we are modifying, then add the new rule.
      // We clear the ingestion block out of the configBuilder entirely to reconstruct it.
      Ingestion.Builder ingestionBuilder = Ingestion.newBuilder();
      if (configBuilder.hasIngestion()) {
        for (IngestionRule rule : configBuilder.getIngestion().getRulesList()) {
          // Keep all existing rules EXCEPT the one targeting DATAPROC
          if (rule.getIntegrationSelector().getIntegration() != Integration.DATAPROC) {
            ingestionBuilder.addRules(rule);
          }
        }
      }
      ingestionBuilder.addRules(enableRule);

      // Update the config builder with the reconstructed ingestion settings.
      configBuilder.setIngestion(ingestionBuilder.build());

      // Build the update request.
      UpdateConfigRequest request = UpdateConfigRequest.newBuilder()
          .setConfig(configBuilder.build())
          .build();

      // Update the config.
      Config response = client.updateConfig(request);
      System.out.printf("Successfully updated config: %s\n", response.getName());
    }
  }
}

Python

from google.api_core.exceptions import NotFound
from google.cloud.datacatalog.lineage import configmanagement_v1

def enable_lineage_ingestion(project_id: str, location: str = "global") -> configmanagement_v1.Config:
    """Enables lineage ingestion for a specific service like Dataproc
    (Managed Service for Apache Spark).

    Args:
        project_id: The ID of your Google Cloud project.
        location: The region location, usually 'global'.

    Returns:
        The updated Configuration object.
    """
    # Initialize client that will be used to send requests.
    client = configmanagement_v1.ConfigManagementServiceClient()

    # The config name format
    name = f"projects/{project_id}/locations/{location}/config"

    try:
        # Retrieve the existing config to preserve other configurations and
        # obtain the latest etag for optimistic concurrency control.
        config = client.get_config(name=name)

        # Filter out existing rules for the integration we are updating
        new_rules = [
            rule for rule in config.ingestion.rules
            if rule.integration_selector.integration != configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
        ]
    except NotFound:
        # If the config does not exist, start fresh
        config = configmanagement_v1.Config(name=name)
        new_rules = []

    # Define the integration to enable tracking for (e.g., DATAPROC).
    integration_selector = configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector(
        integration=configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
    )

    # Set lineage enablement to True to enable tracking.
    lineage_enablement = configmanagement_v1.Config.Ingestion.IngestionRule.LineageEnablement(
        enabled=True
    )

    # Create the ingestion rule.
    enable_rule = configmanagement_v1.Config.Ingestion.IngestionRule(
        integration_selector=integration_selector,
        lineage_enablement=lineage_enablement,
    )

    # Append the new enabling rule and assign it back to the config ingestion rules
    new_rules.append(enable_rule)
    config.ingestion = configmanagement_v1.Config.Ingestion(rules=new_rules)

    # Create the update request using the config (which includes the etag if it existed).
    request = configmanagement_v1.UpdateConfigRequest(
        config=config,
    )

    # Make the request to update the config
    response = client.update_config(request=request)

    print(f"Successfully updated config: {response.name}")
    return response

gcloud

Para ativar a ingestão de linhagem em um serviço específico, use o comando gcloud datalineage config update com uma string JSON inline ou um caminho para um arquivo JSON que define lineageEnablement.enabled como true para o integration específico. As integrações atuais incluem o Serviço Gerenciado para Apache Spark, o BigQuery e o Managed Airflow.

O exemplo a seguir mostra como ativar a ingestão de linhagem de um serviço para um projeto usando uma string JSON inline:

gcloud datalineage config update --project=PROJECT_ID \
  --config='{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "INTEGRATION"
          },
          "lineageEnablement": {
            "enabled": true
          }
        }
      ]
    },
    "etag": "ETAG"
  }'

Substitua:

  • PROJECT_ID: o ID do projeto cuja configuração você quer atualizar.
  • INTEGRATION: a integração para a qual você definiu a configuração (por exemplo, DATAPROC ou BIGQUERY).
  • ETAG: o valor etag retornado de uma solicitação get recente no corpo da solicitação, usado para verificar se a configuração não mudou desde sua última solicitação de leitura.

Para atualizar a configuração usando um arquivo JSON, execute:

gcloud datalineage config update --project=PROJECT_ID --config=CONFIG_FILE

Substitua:

  • CONFIG_FILE: o caminho para o arquivo JSON que contém a configuração.

Para ativar a ingestão de linhagem de um serviço em uma pasta ou organização, substitua --project=PROJECT_ID por um dos seguintes:

  • --folder=FOLDER_ID se quiser atualizar as configurações de ingestão de dados de uma pasta.
  • --organization=ORGANIZATION_ID se você quiser atualizar as configurações de ingestão de dados de uma organização.

REST

Para ativar a ingestão de linhagem em um serviço específico, use o método projects.locations.config.patch com uma regra de ingestão que define lineageEnablement.enabled como true para o integration específico.

Para evitar a substituição acidental de configurações feitas por outros usuários em cenários de leitura-modificação-gravação, inclua o campo etag no corpo da solicitação. Para mais informações, consulte Receber a configuração atual.

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

  • CLIENT_PROJECT_ID: o ID do projeto do cliente usado para faturamento ou cotas.
  • PROJECT_ID: o ID do projeto cuja configuração você quer atualizar.
  • ETAG: o valor etag retornado de uma solicitação get recente.
  • INTEGRATION: o integration para o qual você definiu a configuração. Por exemplo, DATAPROC.

Método HTTP e URL: 

PATCH https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

Corpo JSON da solicitação:

{
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "ETAG"
}

Para enviar a solicitação, expanda uma destas opções:

Você receberá uma resposta JSON semelhante a esta:

{
  "name": "projects/PROJECT_ID/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Para ativar a ingestão de linhagem de um serviço em uma pasta ou organização, substitua projects/PROJECT_ID por folders/FOLDER_ID ou organizations/ORGANIZATION_ID.

Configurar a ingestão de linhagem para vários serviços

Para configurar a ingestão de linhagem de várias integrações simultaneamente, use os métodos projects.locations.config.patch, folders.locations.config.patch ou organizations.locations.config.patch. É possível atualizar a configuração no nível do projeto, da pasta ou da organização definindo várias regras no corpo da solicitação. Para mais informações, consulte Como funciona a configuração de ingestão de dados para integrações de vários serviços.

Nível da organização

Configure a organização para ativar a ingestão de linhagem do Serviço Gerenciado para Apache Spark:

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "X-Goog-User-Project: CLIENT_PROJECT_ID" \
  "https://datalineage.googleapis.com/v1/organizations/ORGANIZATION_ID/locations/global/config" \
  -d '{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "DATAPROC"
          },
          "lineageEnablement": {
            "enabled": true
          }
        }
      ]
    },
    "etag": "ORGANIZATION_CONFIG_ETAG"
  }'

Substitua:

  • ORGANIZATION_ID: o ID da organização cuja configuração você quer atualizar.
  • CLIENT_PROJECT_ID: o ID do seu projeto cliente usado para faturamento ou cotas.
  • ORGANIZATION_CONFIG_ETAG: o valor etag retornado de uma solicitação get recente para a configuração da organização.

Nível da pasta

Configure a pasta para Ativar a ingestão de linhagem de dados do BigQuery:

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "X-Goog-User-Project: CLIENT_PROJECT_ID" \
  "https://datalineage.googleapis.com/v1/folders/FOLDER_ID/locations/global/config" \
  -d '{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "BIGQUERY"
          },
          "lineageEnablement": {
            "enabled": true
          }
        }
      ]
    },
    "etag": "FOLDER_CONFIG_ETAG"
  }'

Substitua:

  • FOLDER_ID: o ID da pasta cuja configuração você quer atualizar.
  • CLIENT_PROJECT_ID: o ID do seu projeto cliente usado para faturamento ou cotas.
  • FOLDER_CONFIG_ETAG: o valor etag retornado de uma solicitação get recente para a configuração da pasta.

Nível do projeto

Configure o projeto para desativar a ingestão de linhagem para o BigQuery e ativar a ingestão de linhagem para o Serviço gerenciado para Apache Airflow simultaneamente:

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "X-Goog-User-Project: CLIENT_PROJECT_ID" \
  "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" \
  -d '{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "BIGQUERY"
          },
          "lineageEnablement": {
            "enabled": false
          }
        },
        {
          "integrationSelector": {
            "integration": "MANAGED_AIRFLOW"
          },
          "lineageEnablement": {
            "enabled": true
          }
        }
      ]
    },
    "etag": "PROJECT_CONFIG_ETAG"
  }'

Substitua:

  • PROJECT_ID: o ID do projeto cuja configuração você quer atualizar.
  • CLIENT_PROJECT_ID: o ID do seu projeto cliente usado para faturamento ou cotas.
  • PROJECT_CONFIG_ETAG: o valor etag retornado de uma solicitação get recente para a configuração do projeto.

A seguir