サービスのデータ リネージの取り込みを構成する

プロジェクト、フォルダ、組織レベルで、特定の統合のリネージ取り込みを選択的に有効または無効にできます。

サポートされている統合と構成シナリオの詳細については、データリネージの取り込みを制御するをご覧ください。

前提条件

リネージの取り込みを制御するには、Data Lineage API を使用する必要があります。Data Lineage API はクライアント ベースの API であるため、課金と割り当て用にクライアント プロジェクトが構成されていることを確認します。

ロールと権限

データ リネージの取り込みを構成して制御するために必要な権限を取得するには、次の Identity and Access Management（IAM）ロールを付与するよう管理者に依頼してください。

データ リネージの取り込みを構成して制御するために必要な権限を取得するには、プロジェクトに対する次の IAM ロールを付与するよう管理者に依頼してください。

• データリネージを構成して管理する: データリネージ管理者 （roles/datalineage.admin）

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

これらの事前定義ロールには、データ リネージの取り込みを構成して制御するために必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

データ リネージの取り込みを構成して制御するには、次の権限が必要です。

• リネージ構成を取得して更新する:
  • datalineage.configs.get
  • datalineage.configs.update

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

1. クライアント プロジェクトで datalineage.googleapis.com API を有効にします。詳細については、データ リネージを有効にするをご覧ください。

2. クライアント プロジェクトを設定します。次の例では、X-Goog-User-Project ヘッダーを使用します。詳細については、システム パラメータをご覧ください。

現在の構成を取得する

リソースでリネージの取り込みが有効になっているかどうかを確認する場合や、構成を変更する前に etag 値を取得する場合は、現在の構成を取得します。

C#

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

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

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

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 メソッドと URL:

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

特定のサービスの系統取り込みを無効にするには、インライン JSON 文字列または特定の integration の lineageEnablement.enabled を false に設定する JSON ファイルのパスを指定して、gcloud datalineage config update コマンドを使用します。

次の例は、インライン 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

特定のサービスのリネージ取り込みを無効にするには、特定の integration の lineageEnablement.enabled を false に設定する取り込みルールを使用して、projects.locations.config.patch メソッドを使用します。

読み取り / 変更 / 書き込みシナリオで他のユーザーが行った構成が誤って上書きされないようにするには、リクエスト本文に etag フィールドを含めます。詳細については、現在の構成を取得するをご覧ください。

リクエストのデータを使用する前に、次のように置き換えます。

• CLIENT_PROJECT_ID: 課金または割り当てに使用されるクライアント プロジェクトの ID。
• PROJECT_ID: 構成を更新するプロジェクトの ID。
• ETAG: 最近の get リクエストから返された etag 値。
• INTEGRATION: 構成を設定した integration。例: DATAPROC

HTTP メソッドと URL:

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）

注: 次のコマンドは、gcloud init または gcloud auth login を実行して、ユーザー アカウントで gcloud CLI にログインしているか、Cloud Shell を使用して自動的に gcloud CLI にログインしていることを前提としています。gcloud auth list を実行すると、現在アクティブなアカウントを確認できます。

リクエスト本文を 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

特定のサービスの系統の取り込みを有効にするには、インライン JSON 文字列または JSON ファイルのパスを指定して gcloud datalineage config update コマンドを使用します。この 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

特定のサービスの系統取り込みを有効にするには、特定の integration の lineageEnablement.enabled を true に設定する取り込みルールを使用して、projects.locations.config.patch メソッドを使用します。

読み取り / 変更 / 書き込みシナリオで他のユーザーが行った構成が誤って上書きされないようにするには、リクエスト本文に etag フィールドを含めます。詳細については、現在の構成を取得するをご覧ください。

リクエストのデータを使用する前に、次のように置き換えます。

• CLIENT_PROJECT_ID: 課金または割り当てに使用されるクライアント プロジェクトの ID。
• PROJECT_ID: 構成を更新するプロジェクトの ID。
• ETAG: 最近の get リクエストから返された etag 値。
• INTEGRATION: 構成を設定した integration。例: DATAPROC

HTTP メソッドと URL:

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）

注: 次のコマンドは、gcloud init または gcloud auth login を実行して、ユーザー アカウントで gcloud CLI にログインしているか、Cloud Shell を使用して自動的に gcloud CLI にログインしていることを前提としています。gcloud auth list を実行すると、現在アクティブなアカウントを確認できます。

リクエスト本文を 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 値。

次のステップ

• データリネージの取り込みを制御する方法について学習する。
• リネージ情報を表示する方法を確認する。