החל מ-10 באפריל 2026, Dataplex Universal Catalog נקרא עכשיו Knowledge Catalog. השמות של ה-API, ספריית הלקוח, ה-CLI ו-IAM לא משתנים. מידע נוסף זמין במאמר הכרות עם קטלוג הידע של Google Cloud.

‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

חיפוש שושלת נתונים בכמה אזורים באמצעות אוטומציה בצד השרת

באמצעות ה-API של searchLineageStreaming אתם יכולים לחפש שושלת נתונים בכמה רמות ובכמה אזורים, ולקבל תרשים מאוחד כסטרימינג בזמן אמת. אתם יכולים ליצור כלים שעוברים על ארכיטקטורות נתונים רחבות, עמוקות או מרובות פרויקטים בלי זמן קצוב לתפוגה של בקשות. מידע נוסף מופיע במאמר מידע על חיפוש שושלת נתונים במספר אזורים.

יכולות עיקריות

ה-API של searchLineageStreaming כולל את היכולות הבאות:

חיפוש לרוחב: עובר על תרשים השושלת שכבה אחר שכבה, במעלה או במורד הזרם, מקבוצה מוגדרת של ישויות שורש, ומחשב במדויק את העומק של כל נכס מקושר.
תשובה בסטרימינג: המערכת מספקת תשובות בזמן אמת, בחלקים, שמכילות תת-תרשימים וקישורים לשושלת היוחסין כשהמערכת בקצה העורפי מגלה אותם. בניגוד לממשקי API רגילים לחיפוש שושלת, שעלולים להגיע לזמן קצוב לתפוגה בתרשימים גדולים עם כמה פרויקטים, הסטרימינג מונע את הזמן הקצוב לתפוגה של בקשות בתרשימים רחבים או עמוקים.
מעבר בין כמה מיקומים וכמה פרויקטים: למרות שאתם מציינים רק פרויקט אחד לחיוב בנתיב הבקשה, ה-API מגלה ועובר באופן אוטומטי על קישורי שושלת היוחסין בכמה פרויקטים ומיקומים גיאוגרפיים ב- Google Cloud , בתנאי שיש לכם את ההרשאות הנדרשות.
שושלת יוחסין מפורטת ברמת העמודה: תמיכה בחיפוש של יחסי תלות ברמת העמודה בין נכסים.
חיפושים עם תו כללי לחיפוש: מאפשרים לאחזר את כל שושלת היוחסין ברמת העמודה של ישות ספציפית, על ידי הוספת הסיומת * לשם המלא (FQN).
תובנות לגבי צינורות עיבוד נתונים: אפשר לאחזר מטא-נתונים לגבי צינורות עיבוד הנתונים (תהליכים) שיצרו את הקישורים של שושלת היוחסין.

לפני שמתחילים

לפני ששולחים בקשות ל-API, צריך לוודא שעומדים בדרישות המוקדמות הבאות בנושאי אבטחה וסביבה:

התפקידים הנדרשים

כדי לקבל את ההרשאות שדרושות לחיפוש קישורים של שושלת נתונים, בקשו מהאדמין להקצות לכם את התפקיד צפייה בשושלת נתונים (roles/datalineage.viewer) ב-IAM בפרויקטים שבהם מאוחסנים הקישורים והתהליכים של שושלת הנתונים. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

זהו תפקיד מוגדר מראש שכולל את ההרשאות שנדרשות כדי לחפש קישורים לשושלת נתונים. כדי לראות בדיוק אילו הרשאות נדרשות, אפשר להרחיב את הקטע ההרשאות הנדרשות:

ההרשאות הנדרשות

כדי לחפש קישורים של שושלת נתונים, נדרשות ההרשאות הבאות:

חיפוש שושלת ברמת הישות: datalineage.events.get בפרויקט שבו הקישור מאוחסן
חיפוש שושלת יוחסין ברמת העמודה: datalineage.events.getFields בפרויקט שבו הקישור מאוחסן
אחזור פרטים מלאים על תהליך צינור עיבוד הנתונים: datalineage.processes.get בפרויקט שבו התהליך מאוחסן

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

היקף המשאבים

כשמגדירים את בקשת ה-API, צריך להבחין בין המשאב שמשמש לחיוב אדמיניסטרטיבי לבין המיקומים בפועל שנסרקים על ידי ה-API:

נתיב ההורה לחיוב: הנתיב parent בבקשת ה-URL צריך להיות בפורמט projects/project/locations/location. הצמד הספציפי הזה של פרויקט ומיקום משמש רק להערכת מכסות החיוב ומגבלות הקצב של יצירת בקשות ל-API.
מיקומי יעד גיאוגרפיים: מגדירים במפורש את האזורים שרוצים שה-backend יסרוק במערך locations בתוך גוף הבקשה.

הגדרת אימות

מאתחלים משתנה סביבה עם אסימון גישה של Google Cloud כדי לאמת את פקודות ה-curl:

export ACCESS_TOKEN=$(gcloud auth print-access-token)

דוגמאות לשימוש

בדוגמאות הבאות נעשה שימוש בנקודת הקצה datalineage.googleapis.com.

חיפוש שושלת נתונים בכמה רמות ובכמה פרויקטים

כדי לבצע חיפוש מעמיק של שושלת היוחסין שעובר בכמה עומקים של התרשים וסורק פרויקטים שונים ב- Google Cloud , מגדירים את המשתנים הבאים:

מגדירים את limits.maxDepth לעומק המעבר הרצוי (מקבל ערכים מ-1 עד 100).
מאכלסים את המערך locations באזורי היעד שרוצים שה-backend יצליב נתונים (לדוגמה, ["us", "us-east1"]).

C#‎

C#

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C# במדריך למתחילים בנושא שושלת נתונים באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של Data Lineage C# API.

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

using Google.Api.Gax.Grpc;
using Google.Api.Gax.ResourceNames;
using Google.Cloud.DataCatalog.Lineage.V1;
using System.Threading.Tasks;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for SearchLineageStreaming</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 async Task SearchLineageStreamingRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        SearchLineageStreamingRequest request = new SearchLineageStreamingRequest
        {
            ParentAsLocationName = LocationName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
            Locations = { "", },
            RootCriteria = new SearchLineageStreamingRequest.Types.RootCriteria(),
            Direction = SearchLineageStreamingRequest.Types.SearchDirection.Unspecified,
            Filters = new SearchLineageStreamingRequest.Types.SearchFilters(),
            Limits = new SearchLineageStreamingRequest.Types.SearchLimits(),
        };
        // Make the request, returning a streaming response
        using LineageClient.SearchLineageStreamingStream response = lineageClient.SearchLineageStreaming(request);

        // Read streaming responses from server until complete
        // Note that C# 8 code can use await foreach
        AsyncResponseStream<SearchLineageStreamingResponse> responseStream = response.GetResponseStream();
        while (await responseStream.MoveNextAsync())
        {
            SearchLineageStreamingResponse responseItem = responseStream.Current;
            // Do something with streamed response
        }
        // The response stream has completed
    }
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במדריך למתחילים בנושא שושלת נתונים באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של Data Lineage Java API.

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

import com.google.api.gax.rpc.ServerStream;
import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.LocationName;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingRequest;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingResponse;
import java.util.ArrayList;

public class AsyncSearchLineageStreaming {

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

  public static void asyncSearchLineageStreaming() 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 (LineageClient lineageClient = LineageClient.create()) {
      SearchLineageStreamingRequest request =
          SearchLineageStreamingRequest.newBuilder()
              .setParent(LocationName.of("[PROJECT]", "[LOCATION]").toString())
              .addAllLocations(new ArrayList<String>())
              .setRootCriteria(SearchLineageStreamingRequest.RootCriteria.newBuilder().build())
              .setFilters(SearchLineageStreamingRequest.SearchFilters.newBuilder().build())
              .setLimits(SearchLineageStreamingRequest.SearchLimits.newBuilder().build())
              .build();
      ServerStream<SearchLineageStreamingResponse> stream =
          lineageClient.searchLineageStreamingCallable().call(request);
      for (SearchLineageStreamingResponse response : stream) {
        // Do something when a response is received.
      }
    }
  }
}

Node.js

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במדריך למתחילים בנושא שושלת נתונים באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של Data Lineage Node.js API.

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

/**
 * 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.
 * TODO(developer): Uncomment these variables before running the sample.
 */
/**
 *  Required. The project and location to initiate the search from.
 */
// const parent = 'abc123'
/**
 *  Required. The locations to search in.
 */
// const locations = ['abc','def']
/**
 *  Required. Criteria for the root of the search.
 */
// const rootCriteria = {}
/**
 *  Required. Direction of the search.
 */
// const direction = {}
/**
 *  Optional. Filters for the search.
 */
// const filters = {}
/**
 *  Optional. Limits for the search.
 */
// const limits = {}

// Imports the Lineage library
const {LineageClient} = require('@google-cloud/lineage').v1;

// Instantiates a client
const lineageClient = new LineageClient();

async function callSearchLineageStreaming() {
  // Construct request
  const request = {
    parent,
    locations,
    rootCriteria,
    direction,
  };

  // Run request
  const stream = await lineageClient.searchLineageStreaming(request);
  stream.on('data', (response) => { console.log(response) });
  stream.on('error', (err) => { throw(err) });
  stream.on('end', () => { /* API call completed */ });
}

callSearchLineageStreaming();

Python

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Python במדריך למתחילים בנושא שושלת נתונים באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של Data Lineage Python API.

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

# 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_v1


def sample_search_lineage_streaming():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.SearchLineageStreamingRequest(
        parent="parent_value",
        locations=["locations_value1", "locations_value2"],
        direction="UPSTREAM",
    )

    # Make the request
    stream = client.search_lineage_streaming(request=request)

    # Handle the response
    for response in stream:
        print(response)

Ruby

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Ruby במדריך למתחילים בנושא שושלת נתונים באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של Data Lineage Ruby API.

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

require "google/cloud/data_catalog/lineage/v1"

##
# Snippet for the search_lineage_streaming call in the Lineage service
#
# 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/ruby/docs/reference.
#
# This is an auto-generated example demonstrating basic usage of
# Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client#search_lineage_streaming.
#
def search_lineage_streaming
  # Create a client object. The client can be reused for multiple calls.
  client = Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client.new

  # Create a request. To set request fields, pass in keyword arguments.
  request = Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingRequest.new

  # Call the search_lineage_streaming method to start streaming.
  output = client.search_lineage_streaming request

  # The returned object is a streamed enumerable yielding elements of type
  # ::Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingResponse
  output.each do |current_response|
    p current_response
  end
end

REST

כדי לחפש שושלת נתונים, משתמשים בשיטה searchLineageStreaming.

לפני שמשתמשים בנתוני הבקשה, צריך להחליף את הנתונים הבאים:

‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud שמשמש לחיוב אדמיניסטרטיבי ולהערכת מכסות.
‫LOCATION_ID: המיקום ב- Google Cloud , למשל us-central1.
‫SOURCE_PROJECT_ID: Google Cloud מזהה הפרויקט שבו נמצאת טבלת המקור.
‫DATASET_ID: מזהה מערך הנתונים ב-BigQuery.
‫TABLE_ID: מזהה הטבלה ב-BigQuery.

ה-method של ה-HTTP וכתובת ה-URL:

POST https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming

תוכן בקשת JSON:

{
  "parent": "projects/PROJECT_ID/locations/LOCATION_ID",
  "locations": [
    "LOCATION_ID",
    "us-east1",
    "us-central1"
  ],
  "rootCriteria": {
    "entities": {
      "entities": [
        {
          "fullyQualifiedName": "bigquery:SOURCE_PROJECT_ID.DATASET_ID.TABLE_ID"
        }
      ]
    }
  },
  "direction": "DOWNSTREAM",
  "limits": {
    "maxDepth": 10,
    "maxResults": 5000
  }
}

כדי לשלוח את הבקשה צריך להרחיב אחת מהאפשרויות הבאות:

‫Curl (Linux,‏ macOS או Cloud Shell)

הערה: הפקודה הבאה מבוססת על ההנחה שנכנסתם ל-CLI של gcloud באמצעות חשבון המשתמש שלכם, על ידי הרצת gcloud init או gcloud auth login, או באמצעות Cloud Shell שמחבר אתכם אוטומטית ל-CLI של gcloud. כדי לבדוק איזה חשבון פעיל, אפשר להריץ את הפקודה gcloud auth list.

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming"

‎PowerShell (Windows)

הערה: הפקודה הבאה מבוססת על ההנחה שנכנסתם ל-CLI של gcloud באמצעות חשבון המשתמש שלכם, על ידי הרצת gcloud init או gcloud auth login. כדי לבדוק איזה חשבון פעיל, אפשר להריץ את הפקודה gcloud auth list.

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

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

אתם אמורים לקבל תגובת JSON שדומה לזו:

{
  "links": [
    {
      "source": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.source_table"
      },
      "target": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.target_table"
      },
      "depth": 1,
      "location": "us"
    }
  ]
}

חיפוש בכמה מיקומים גיאוגרפיים

כדי להגביל או להרחיב את הסריקה של תרשים השושלת, אפשר לשנות את האזורים הגיאוגרפיים שמועברים בתוך השדה החוזר (repeated) של המערך locations.

C#‎

C#

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

using Google.Api.Gax.Grpc;
using Google.Api.Gax.ResourceNames;
using Google.Cloud.DataCatalog.Lineage.V1;
using System.Threading.Tasks;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for SearchLineageStreaming</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 async Task SearchLineageStreamingRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        SearchLineageStreamingRequest request = new SearchLineageStreamingRequest
        {
            ParentAsLocationName = LocationName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
            Locations = { "", },
            RootCriteria = new SearchLineageStreamingRequest.Types.RootCriteria(),
            Direction = SearchLineageStreamingRequest.Types.SearchDirection.Unspecified,
            Filters = new SearchLineageStreamingRequest.Types.SearchFilters(),
            Limits = new SearchLineageStreamingRequest.Types.SearchLimits(),
        };
        // Make the request, returning a streaming response
        using LineageClient.SearchLineageStreamingStream response = lineageClient.SearchLineageStreaming(request);

        // Read streaming responses from server until complete
        // Note that C# 8 code can use await foreach
        AsyncResponseStream<SearchLineageStreamingResponse> responseStream = response.GetResponseStream();
        while (await responseStream.MoveNextAsync())
        {
            SearchLineageStreamingResponse responseItem = responseStream.Current;
            // Do something with streamed response
        }
        // The response stream has completed
    }
}

Java

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

import com.google.api.gax.rpc.ServerStream;
import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.LocationName;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingRequest;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingResponse;
import java.util.ArrayList;

public class AsyncSearchLineageStreaming {

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

  public static void asyncSearchLineageStreaming() 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 (LineageClient lineageClient = LineageClient.create()) {
      SearchLineageStreamingRequest request =
          SearchLineageStreamingRequest.newBuilder()
              .setParent(LocationName.of("[PROJECT]", "[LOCATION]").toString())
              .addAllLocations(new ArrayList<String>())
              .setRootCriteria(SearchLineageStreamingRequest.RootCriteria.newBuilder().build())
              .setFilters(SearchLineageStreamingRequest.SearchFilters.newBuilder().build())
              .setLimits(SearchLineageStreamingRequest.SearchLimits.newBuilder().build())
              .build();
      ServerStream<SearchLineageStreamingResponse> stream =
          lineageClient.searchLineageStreamingCallable().call(request);
      for (SearchLineageStreamingResponse response : stream) {
        // Do something when a response is received.
      }
    }
  }
}

Node.js

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במדריך למתחילים בנושא שושלת נתונים באמצעות ספריות לקוח.

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

/**
 * 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.
 * TODO(developer): Uncomment these variables before running the sample.
 */
/**
 *  Required. The project and location to initiate the search from.
 */
// const parent = 'abc123'
/**
 *  Required. The locations to search in.
 */
// const locations = ['abc','def']
/**
 *  Required. Criteria for the root of the search.
 */
// const rootCriteria = {}
/**
 *  Required. Direction of the search.
 */
// const direction = {}
/**
 *  Optional. Filters for the search.
 */
// const filters = {}
/**
 *  Optional. Limits for the search.
 */
// const limits = {}

// Imports the Lineage library
const {LineageClient} = require('@google-cloud/lineage').v1;

// Instantiates a client
const lineageClient = new LineageClient();

async function callSearchLineageStreaming() {
  // Construct request
  const request = {
    parent,
    locations,
    rootCriteria,
    direction,
  };

  // Run request
  const stream = await lineageClient.searchLineageStreaming(request);
  stream.on('data', (response) => { console.log(response) });
  stream.on('error', (err) => { throw(err) });
  stream.on('end', () => { /* API call completed */ });
}

callSearchLineageStreaming();

Python

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

# 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_v1


def sample_search_lineage_streaming():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.SearchLineageStreamingRequest(
        parent="parent_value",
        locations=["locations_value1", "locations_value2"],
        direction="UPSTREAM",
    )

    # Make the request
    stream = client.search_lineage_streaming(request=request)

    # Handle the response
    for response in stream:
        print(response)

Ruby

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

require "google/cloud/data_catalog/lineage/v1"

##
# Snippet for the search_lineage_streaming call in the Lineage service
#
# 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/ruby/docs/reference.
#
# This is an auto-generated example demonstrating basic usage of
# Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client#search_lineage_streaming.
#
def search_lineage_streaming
  # Create a client object. The client can be reused for multiple calls.
  client = Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client.new

  # Create a request. To set request fields, pass in keyword arguments.
  request = Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingRequest.new

  # Call the search_lineage_streaming method to start streaming.
  output = client.search_lineage_streaming request

  # The returned object is a streamed enumerable yielding elements of type
  # ::Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingResponse
  output.each do |current_response|
    p current_response
  end
end

REST

לפני שמשתמשים בנתוני הבקשה, צריך להחליף את הנתונים הבאים:

‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud שמשמש לחיוב אדמיניסטרטיבי ולהערכת מכסות.
‫LOCATION_ID: המיקום ב- Google Cloud , למשל us-central1.
‫SOURCE_PROJECT_ID: Google Cloud מזהה הפרויקט שבו נמצאת טבלת המקור.
‫DATASET_ID: מזהה מערך הנתונים ב-BigQuery.
‫TABLE_ID: מזהה הטבלה ב-BigQuery.

ה-method של ה-HTTP וכתובת ה-URL:

POST https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming

תוכן בקשת JSON:

{
  "parent": "projects/PROJECT_ID/locations/LOCATION_ID",
  "locations": [
    "LOCATION_ID",
    "us-east1",
    "us-central1"
  ],
  "rootCriteria": {
    "entities": {
      "entities": [
        {
          "fullyQualifiedName": "bigquery:SOURCE_PROJECT_ID.DATASET_ID.TABLE_ID"
        }
      ]
    }
  },
  "direction": "DOWNSTREAM",
  "limits": {
    "maxDepth": 10,
    "maxResults": 5000
  }
}

כדי לשלוח את הבקשה צריך להרחיב אחת מהאפשרויות הבאות:

‫Curl (Linux,‏ macOS או Cloud Shell)

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming"

‎PowerShell (Windows)

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

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

אתם אמורים לקבל תגובת JSON שדומה לזו:

{
  "links": [
    {
      "source": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.source_table"
      },
      "target": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.target_table"
      },
      "depth": 1,
      "location": "us"
    }
  ]
}

אחזור שמות של תהליכים לקישורי שושלת נתונים

כברירת מחדל, ה-API משמיט את פרטי התהליך (maxProcessPerLink מוגדר כברירת מחדל ל-0). כדי לאחזר את שמות המשאבים של צינורות עיבוד הנתונים שיצרו את קישורי הנתונים, מגדירים את limits.maxProcessPerLink למספר שלם חיובי שאינו אפס.

לדוגמה:

Java

כדי לבצע אימות ב-Data Lineage, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.Process;
import com.google.cloud.datacatalog.lineage.v1.ProcessName;

public class SyncGetProcessProcessname {

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

  public static void syncGetProcessProcessname() 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 (LineageClient lineageClient = LineageClient.create()) {
      ProcessName name = ProcessName.of("[PROJECT]", "[LOCATION]", "[PROCESS]");
      Process response = lineageClient.getProcess(name);
    }
  }
}

REST

לפני שמשתמשים בנתוני הבקשה, צריך להחליף את הנתונים הבאים:

‫BILLING_PROJECT_ID: הפרויקט ב- Google Cloudשמשמש לחיוב אדמיניסטרטיבי ולהערכת מכסות. מזינים את מספר הפרויקט או את מזהה הפרויקט.
‫LOCATION_ID: המיקום Google Cloud שבו מתבצע החיפוש של שושלת היוחסין.
‫FULLY_QUALIFIED_NAME: השם המלא (FQN) של ישות היעד, בפורמט bigquery:PROJECT_ID.DATASET_ID.TABLE_ID.

ה-method של ה-HTTP וכתובת ה-URL:

POST https://datalineage.googleapis.com/v1/projects/BILLING_PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming

תוכן בקשת JSON:

{
  "parent": "projects/BILLING_PROJECT_ID/locations/LOCATION_ID",
  "locations": [
    "LOCATION_ID"
  ],
  "rootCriteria": {
    "entities": {
      "entities": [
        {
          "fullyQualifiedName": "FULLY_QUALIFIED_NAME"
        }
      ]
    }
  },
  "direction": "UPSTREAM",
  "limits": {
    "maxProcessPerLink": 5
  }
}

כדי לשלוח את הבקשה צריך להרחיב אחת מהאפשרויות הבאות:

‫Curl (Linux,‏ macOS או Cloud Shell)

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/BILLING_PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming"

‎PowerShell (Windows)

שומרים את גוף הבקשה בקובץ בשם request.json ומריצים את הפקודה הבאה:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

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

התנהגות התגובה: הסטרימינג שנוצר מאכלס את השדה links[].processes בהודעות תהליך שמכילות רק את שם המשאב המוחלט של המערכת (למשל projects/my-project/locations/us/processes/my-process).

אחזור פרטי תהליך מלאים באמצעות FieldMask

אם אתם צריכים מטא-נתונים מבניים מלאים על צינור עיבוד נתונים (כמו displayName, מערכת attributes או הפעלה origin) במקום רק את שם המשאב, אתם צריכים להשתמש ב-API FieldMask:

צריך להזין ערך שאינו אפס במאפיין limits.maxProcessPerLink.
מוסיפים את הפרמטר fields של השאילתה לנתיב כתובת ה-URL, ומציינים את הערך links.processes.process יחד עם שדות חובה אחרים.

לדוגמה:

curl -H "Authorization: Bearer ${ACCESS_TOKEN}" \
-H "Content-Type: application/json" \
-X POST "https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming?fields=links.processes.process,links.source,links.target,links.depth" \
--data '{
  "parent": "projects/my-billing-project/locations/us",
  "locations": ["us"],
  "rootCriteria": {
    "entities": {
      "entities": [{
        "fullyQualifiedName": "bigquery:my-project.dataset.target_table"
      }]
    }
  },
  "direction": "UPSTREAM",
  "limits": {
    "maxProcessPerLink": 5
  }
}'

חיפוש שושלת נתונים ברמת הטבלה וברמת העמודה

אפשר לחפש שושלת יוחסין גם ברמת הטבלה (ברמת הנכס) וגם ברמת העמודה (ברמת השדה) בבקשה אחת, על ידי ציון כמה ישויות ברשימה rootCriteria.entities.entities:

למעקב ברמת הטבלה, משמיטים את המערך field.
לשושלת ברמת העמודה, מציינים עמודה אחת במערך field.

לדוגמה:

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [
             {
               "fullyQualifiedName": "bigquery:my-project.dataset.table_a"
             },
             {
               "fullyQualifiedName": "bigquery:my-project.dataset.table_b",
               "field": ["email"]
             }
           ]
         }
       },
       "direction": "DOWNSTREAM"
     }'

שימוש בתווים כלליים לחיפוש שושלת ברמת העמודה

כדי לחפש את כל שושלות היוחסין הזמינות ברמת העמודה בטבלה ספציפית, בלי לרשום כל עמודה בנפרד, משתמשים בתו הכללי לחיפוש * כערך היחיד במערך field.

לדוגמה:

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table",
             "field": ["*"]
           }]
         }
       },
       "direction": "DOWNSTREAM"
     }'

סינון תוצאות של שושלת נתונים

כדי לצמצם את תוצאות החיפוש של שושלת היוחסין, אפשר להשתמש בבלוק filters בגוף הבקשה.

סינון לפי סוג התלות

כדי להגביל את התוצאות לסוגי תלות ספציפיים, כמו עותקים ישירים (EXACT_COPY) או טרנספורמציות כמו סינון וקיבוץ (OTHER), משתמשים במסנן dependencyTypes.

לדוגמה:

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table"
           }]
         }
       },
       "direction": "DOWNSTREAM",
       "filters": {
         "dependencyTypes": ["EXACT_COPY"]
       }
     }'

החרגת שושלת נתונים ברמת העמודה (חיפוש בטבלה בלבד)

כדי לוודא שהחיפוש יחזיר רק שושלת יוחסין ברמת הטבלה ולא יכלול שושלת יוחסין ברמת העמודה, מגדירים את המסנן entitySet לערך ENTITIES.

לדוגמה:

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table"
           }]
         }
       },
       "direction": "DOWNSTREAM",
       "filters": {
         "entitySet": "ENTITIES"
       }
     }'

סינון לפי טווח תאריכים

אתם יכולים להגביל את תוצאות החיפוש של שושלת היוחסין למרווח זמן ספציפי.

לדוגמה, כדי לחפש נתוני שושלת שנוצרו אחרי חותמת זמן ספציפית, משתמשים בבקשה הבאה:

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table"
           }]
         }
       },
       "direction": "DOWNSTREAM",
       "filters": {
         "timeRange": {
           "startTime": "2026-01-01T00:00:00Z"
         }
       }
     }'

פתרון בעיות: טיפול במיקומים שלא ניתן להגיע אליהם ובגרפים חלקיים

מכיוון שה-API לסטרימינג סורק בו-זמנית קבוצה מבוזרת של פרויקטים ומיקומים, יכול להיות שבמהלך ההפעלה חלק מהאזורים המרוחקים לא יהיו זמינים באופן זמני, לא יתקשרו או יוגדרו באופן שגוי.

הבעיה: הפריסה של תרשים שושלת היוחסין שמוחזר נראית לא מלאה או שחסרים בה דילוגים אזוריים צפויים.
אבחון: כדי להגן על תקינות הנתונים, הסטרימינג של searchLineageStreamingResponse מאכלס שדה unreachable ייעודי (מחרוזת חוזרת) במיקומים בעייתיים, בפורמט projects/PROJECT_NUMBER/locations/LOCATION (לדוגמה, projects/123456789/locations/us-east1).
שיטה מומלצת: תמיד כדאי לבנות את אפליקציות הלקוח כך שיבדקו את השדה unreachable כדי לוודא שכל הנתונים קיימים לפני עיבוד התרשים.

חיפוש שושלת נתונים בכמה אזורים באמצעות אוטומציה בצד השרת קל לארגן דפים בעזרת אוספים אפשר לשמור ולסווג תוכן על סמך ההעדפות שלך.

יכולות עיקריות

לפני שמתחילים

התפקידים הנדרשים

ההרשאות הנדרשות

היקף המשאבים

הגדרת אימות

דוגמאות לשימוש

חיפוש שושלת נתונים בכמה רמות ובכמה פרויקטים

C#‎

C#

Java

Java

Node.js

Node.js

Python

Python

Ruby

Ruby

REST

‫Curl (Linux,‏ macOS או Cloud Shell)

‎PowerShell (Windows)

חיפוש בכמה מיקומים גיאוגרפיים

C#‎

C#

Java

Java

Node.js

Java

Python

Python

Ruby

Ruby

REST

‫Curl (Linux,‏ macOS או Cloud Shell)

‎PowerShell (Windows)

אחזור שמות של תהליכים לקישורי שושלת נתונים

Java

Java

REST

‫Curl (Linux,‏ macOS או Cloud Shell)

‎PowerShell (Windows)

אחזור פרטי תהליך מלאים באמצעות FieldMask

חיפוש שושלת נתונים ברמת הטבלה וברמת העמודה

שימוש בתווים כלליים לחיפוש שושלת ברמת העמודה

סינון תוצאות של שושלת נתונים

סינון לפי סוג התלות

החרגת שושלת נתונים ברמת העמודה (חיפוש בטבלה בלבד)

סינון לפי טווח תאריכים

פתרון בעיות: טיפול במיקומים שלא ניתן להגיע אליהם ובגרפים חלקיים

המאמרים הבאים

חיפוש שושלת נתונים בכמה אזורים באמצעות אוטומציה בצד השרת