Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

为服务配置数据沿袭提取

您可以在项目、文件夹或组织级层选择性地为特定集成启用或停用沿袭提取。

如需详细了解支持的集成和配置方案，请参阅控制数据沿袭注入。

前提条件

如需控制沿袭数据摄取，您必须使用 Data Lineage API。请确保您已为结算和配额配置客户端项目，因为 Data Lineage API 是基于客户端的 API。

角色与权限

如需获得配置和控制数据沿袭提取所需的权限，请让管理员向您授予以下 Identity and Access Management (IAM) 角色：

如需获得配置和控制数据沿袭提取所需的权限，请让管理员向您授予项目的以下 IAM 角色：

配置和管理数据沿袭：Data Lineage Administrator (roles/datalineage.admin)

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

这些预定义角色包含配置和控制数据沿袭提取所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

如需配置和控制数据沿袭提取，您需要具备以下权限：

获取和更新沿袭配置：
- datalineage.configs.get
- datalineage.configs.update

您也可以使用自定义角色或其他预定义角色来获取这些权限。

在客户端项目中启用 datalineage.googleapis.com API。如需了解详情，请参阅启用数据沿袭。
设置客户端项目。对于以下示例，请使用 X-Goog-User-Project 标头。如需了解详情，请参阅系统参数。

获取当前配置

如需检查是否已为资源启用谱系提取，或在修改配置之前获取 etag 值，请检索当前配置。

C#

试用此示例之前，请按照《Knowledge Catalog 快速入门：使用客户端库》中的 C# 设置说明进行操作。如需了解详情，请参阅 Knowledge Catalog C# API 参考文档。

如需向 Knowledge Catalog 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。

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

试用此示例之前，请按照《Knowledge Catalog 快速入门：使用客户端库》中的 Go 设置说明进行操作。如需了解详情，请参阅 Knowledge Catalog Go API 参考文档。

如需向 Knowledge Catalog 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。


//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

试用此示例之前，请按照《Knowledge Catalog 快速入门：使用客户端库》中的 Java 设置说明进行操作。如需了解详情，请参阅 Knowledge Catalog Java API 参考文档。

如需向 Knowledge Catalog 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。

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

试用此示例之前，请按照《Knowledge Catalog 快速入门：使用客户端库》中的 Python 设置说明进行操作。如需了解详情，请参阅 Knowledge Catalog Python API 参考文档。

如需向 Knowledge Catalog 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。

# 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

如需查看当前谱系配置，请使用 gcloud datalineage config describe 命令。您可以检索项目、文件夹或组织的配置。

以下示例展示了如何获取当前项目的配置：

gcloud datalineage config describe

例如，如需获取特定项目的配置，请使用 --project 标志：

gcloud datalineage config describe --project=PROJECT_ID

替换以下内容：

PROJECT_ID：您要查看其配置的项目的 ID。

如需查看文件夹或组织的服务的当前谱系提取配置，请将 --project=PROJECT_ID 替换为以下内容之一：

--folder=FOLDER_ID（如果您想查看文件夹的数据注入设置）。
--organization=ORGANIZATION_ID（如果您想查看组织的数据注入设置）。

REST

如需查看当前沿袭配置，请使用 projects.locations.config.get 方法。您可以检索项目、文件夹或组织的配置。

以下示例展示了如何获取项目的配置：

在使用任何请求数据之前，请先进行以下替换：

CLIENT_PROJECT_ID：用于结算或配额的客户端项目的 ID。
PROJECT_ID：您要查看其配置的项目的 ID。

HTTP 方法和网址：

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

如需发送您的请求，请展开以下选项之一：

curl（Linux、macOS 或 Cloud Shell）

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI，或者使用了 Cloud Shell，这会使您自动登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

执行以下命令：

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "x-goog-user-project: CLIENT_PROJECT_ID" \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config"

PowerShell (Windows)

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred"; "x-goog-user-project" = "CLIENT_PROJECT_ID" }

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" | Select-Object -Expand Content

该命令会返回以下输出之一：

如果您未提供任何谱系提取设置，则会获得一个包含空 ingestion 对象的输出：
```
{
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {}
}
  
```
这意味着该服务使用默认的沿袭数据提取设置。在此示例中，Managed Service for Apache Spark 的谱系提取设置是 enabled。

如果您明确启用谱系提取，则会获得以下输出：

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

如果沿袭注入已停用，您会看到以下输出：

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

如需获取文件夹或组织的配置，请将 projects/"PROJECT_ID 替换为 folders/FOLDER_ID 或 organizations/ORGANIZATION_ID。

响应中的 etag 字段是服务器根据配置的当前值生成的校验和。使用 patch 方法更新配置时，您可以在请求正文中添加最近一次 get 请求返回的 etag 值。如果您提供 etag，Knowledge Catalog 会使用它来验证自您上次读取请求以来配置是否发生了更改。如果存在不匹配，更新请求会失败。这样可以防止您在读-修改-写场景中无意中覆盖其他用户所做的配置。如果您未在 patch 请求中提供 etag，Knowledge Catalog 会无条件覆盖配置。

为服务停用沿袭提取

如需管理费用、强制执行数据治理政策，或排除无法从沿袭跟踪中受益的开发项目和其他工作负载，请为相应服务停用沿袭提取。

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

如需针对特定服务停用沿袭提取，请使用 gcloud datalineage config update 命令，并提供内嵌 JSON 字符串或 JSON 文件的路径，该字符串或文件会将特定 integration 的 lineageEnablement.enabled 设置为 false。

以下示例展示了如何使用内嵌 JSON 字符串为项目停用服务的谱系提取：

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

替换以下内容：

PROJECT_ID：要更新其配置的项目的 ID。
INTEGRATION：您要为其设置配置的集成。例如 DATAPROC 或 BIGQUERY。
ETAG：请求正文中最近一次 get 请求返回的 etag 值，用于验证自上次读取请求以来配置是否发生了变化。

如需使用 JSON 文件更新配置，请运行以下命令：

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

替换以下内容：

CONFIG_FILE：包含配置的 JSON 文件的路径。

如需为文件夹或组织停用服务的沿袭提取，请将 --project=PROJECT_ID 替换为以下值之一：

如果您想更新文件夹的数据注入设置，请设置为 --folder=FOLDER_ID。
如果您想更新组织的数据注入设置，请使用 --organization=ORGANIZATION_ID。

REST

如需针对特定服务停用沿袭提取，请使用 projects.locations.config.patch 方法，并使用将特定 integration 的 lineageEnablement.enabled 设置为 false 的提取规则。

为防止在读取-修改-写入场景中无意中覆盖其他用户所做的配置，您可以在请求正文中添加 etag 字段。如需了解详情，请参阅获取当前配置。

在使用任何请求数据之前，请先进行以下替换：

CLIENT_PROJECT_ID：用于结算或配额的客户端项目的 ID。
PROJECT_ID：要更新其配置的项目的 ID。
ETAG：最近一次 get 请求返回的 etag 值。
INTEGRATION：您设置配置的 integration。例如 DATAPROC。

HTTP 方法和网址：

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

请求 JSON 正文：

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

如需发送您的请求，请展开以下选项之一：

curl（Linux、macOS 或 Cloud Shell）

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "x-goog-user-project: CLIENT_PROJECT_ID" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config"

PowerShell (Windows)

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred"; "x-goog-user-project" = "CLIENT_PROJECT_ID" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" | Select-Object -Expand Content

您应该收到类似以下内容的 JSON 响应：

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

如需为文件夹或组织停用谱系提取，请将 projects/"PROJECT_ID 替换为 folders/FOLDER_ID 或 organizations/ORGANIZATION_ID。

为服务启用沿袭提取

如需在停用跟踪后恢复跟踪，或启用默认处于停用状态的集成，请为服务启用谱系提取。

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

如需为特定服务启用谱系提取，请使用 gcloud datalineage config update 命令，并提供内嵌 JSON 字符串或 JSON 文件的路径，该文件将特定 integration 的 lineageEnablement.enabled 设置为 true。目前的集成包括 Managed Service for Apache Spark、BigQuery 和 Managed Airflow。

以下示例展示了如何使用内嵌 JSON 字符串为项目启用服务的沿袭提取：

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

替换以下内容：

PROJECT_ID：要更新其配置的项目的 ID。
INTEGRATION：您设置配置的集成（例如 DATAPROC 或 BIGQUERY）。
ETAG：请求正文中最近一次 get 请求返回的 etag 值，用于验证自上次读取请求以来配置是否发生了变化。

如需使用 JSON 文件更新配置，请运行以下命令：

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

替换以下内容：

CONFIG_FILE：包含配置的 JSON 文件的路径。

如需为文件夹或组织启用服务的沿袭提取，请将 --project=PROJECT_ID 替换为以下值之一：

如果您想更新文件夹的数据注入设置，请设置为 --folder=FOLDER_ID。
如果您想更新组织的数据注入设置，请使用 --organization=ORGANIZATION_ID。

REST

如需为特定服务启用沿袭提取，请使用 projects.locations.config.patch 方法，并提供将特定 integration 的 lineageEnablement.enabled 设置为 true 的提取规则。

为防止在读取-修改-写入场景中无意中覆盖其他用户所做的配置，您可以在请求正文中添加 etag 字段。如需了解详情，请参阅获取当前配置。

在使用任何请求数据之前，请先进行以下替换：

CLIENT_PROJECT_ID：用于结算或配额的客户端项目的 ID。
PROJECT_ID：要更新其配置的项目的 ID。
ETAG：最近一次 get 请求返回的 etag 值。
INTEGRATION：您设置配置的 integration。例如 DATAPROC。

HTTP 方法和网址：

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

请求 JSON 正文：

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

如需发送您的请求，请展开以下选项之一：

curl（Linux、macOS 或 Cloud Shell）

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "x-goog-user-project: CLIENT_PROJECT_ID" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config"

PowerShell (Windows)

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred"; "x-goog-user-project" = "CLIENT_PROJECT_ID" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" | Select-Object -Expand Content

您应该收到类似以下内容的 JSON 响应：

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

如需为文件夹或组织启用服务的谱系提取，请将 projects/PROJECT_ID 替换为 folders/FOLDER_ID 或 organizations/ORGANIZATION_ID。

为多个服务配置沿袭提取

如需同时为多个集成配置沿袭提取，请使用 projects.locations.config.patch、folders.locations.config.patch 或 organizations.locations.config.patch 方法。您可以在项目、文件夹或组织级层更新配置，方法是在请求正文中定义多条规则。如需了解详情，请参阅多服务集成的数据注入配置的工作原理。

组织级层

将组织配置为启用 Managed Service for 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"
  }'

替换以下内容：

ORGANIZATION_ID：您要更新其配置的组织的 ID。
CLIENT_PROJECT_ID：用于结算或配额的客户端项目的 ID。
ORGANIZATION_CONFIG_ETAG：最近针对组织配置发出的 get 请求返回的 etag 值。

文件夹级层

将文件夹配置为启用 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"
  }'

替换以下内容：

FOLDER_ID：您要更新其配置的文件夹的 ID。
CLIENT_PROJECT_ID：用于结算或配额的客户端项目的 ID。
FOLDER_CONFIG_ETAG：最近针对文件夹配置发出的 get 请求返回的 etag 值。

项目级

配置项目，以同时停用 BigQuery 的谱系提取功能并启用 Managed Service for Apache Airflow 的谱系提取功能：

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"
  }'

替换以下内容：

PROJECT_ID：要更新其配置的项目的 ID。
CLIENT_PROJECT_ID：用于结算或配额的客户端项目的 ID。
PROJECT_CONFIG_ETAG：最近针对项目配置发出的 get 请求返回的 etag 值。

后续步骤

了解如何控制数据沿袭注入。
了解如何查看沿袭信息。

为服务配置数据沿袭提取 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

前提条件

角色与权限

所需权限

获取当前配置

C#

C#

Go

Go

Java

Java

Python

Python

gcloud

REST

curl（Linux、macOS 或 Cloud Shell）

PowerShell (Windows)

为服务停用沿袭提取

Java

Python

gcloud

REST

curl（Linux、macOS 或 Cloud Shell）

PowerShell (Windows)

为服务启用沿袭提取

Java

Python

gcloud

REST

curl（Linux、macOS 或 Cloud Shell）

PowerShell (Windows)

为多个服务配置沿袭提取

组织级层

文件夹级层

项目级

后续步骤

为服务配置数据沿袭提取