שילוב עם OpenLineage

בעזרת שושלת היוחסין של הנתונים ב-Knowledge Catalog (לשעבר Dataplex Universal Catalog) אתם יכולים לייבא אירועי OpenLineage כדי לעקוב אחרי תנועת הנתונים ממערכות חיצוניות ולהציג אותה באופן חזותי. אתם יכולים לאחד את שושלת היוחסין של צינור עיבוד נתונים בהתאמה אישית עם שושלת היוחסין המובנית משירותיGoogle Cloud בתרשים שושלת יוחסין אחד.

סקירה כללית

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

‫Knowledge Catalog פועל כצרכן של OpenLineage באמצעות Data Lineage API. אתם יכולים לייבא אירועי OpenLineage ממערכות חיצוניות ומצינורות עיבוד נתונים בהתאמה אישית, כדי להציג אותם בממשק האינטרנט של Knowledge Catalog לצד מידע מובנה על שושלת הנתונים משירותי Google Cloud , כמו BigQuery, ‏ Managed Service for Apache Airflow, ‏ Cloud Data Fusion ו-Managed Service for Apache Spark.

כדי לייבא אירועים של OpenLineage שמשתמשים במפרט של OpenLineage, משתמשים ב-method של ה-API בארכיטקטורת REST‏ ProcessOpenLineageRunEvent וממפים את ההיבטים של OpenLineage למאפיינים של Data Lineage API.

מגבלות השילוב של OpenLineage

  • גרסאות נתמכות: ה-Data Lineage API תומך בגרסאות העיקריות 1 ו-2 של OpenLineage.

  • פעולות API: נקודת הקצה ל-API של Data Lineage‏ ProcessOpenLineageRunEvent פועלת רק כצרכן של הודעות OpenLineage, ולא כיצרן. באמצעות ה-API אפשר לשלוח ל-Knowledge Catalog מידע על שושלת נתונים שנוצר על ידי כל כלי או מערכת שתואמים ל-OpenLineage. בחלק משירותי Google Cloud , כמו Managed Service for Apache Spark ו-Managed Airflow, יש יצרני OpenLineage מובְנים שיכולים לשלוח אירועים לנקודת הקצה הזו, וכך להפוך את תיעוד שושלת היוחסין מהשירותים האלה לאוטומטי.

  • תכונות שלא נתמכות: Data Lineage API לא תומך בתכונות הבאות:

    • כל גרסה עתידית של OpenLineage עם שינויים בפורמט ההודעות
    • DatasetEvent
    • JobEvent

  • גודל ההודעה: הגודל המקסימלי של הודעה בודדת הוא 5MB.

  • אורך השם: האורך של כל שם מלא בקלט ובפלט מוגבל ל-4,000 תווים.

  • שושלת ברמת העמודה: נקודת הקצה ProcessOpenLineageRunEvent יוצרת קצוות של שושלת ברמת הטבלה רק לאירועים מותאמים אישית של OpenLineage. תרשימי שושלת יוחסין ברמת העמודה נוצרים רק בשביל טרנספורמציות SQL ב-BigQuery ומשימות של Managed Service for Apache Spark. היבטים ברמת העמודה בהתאמה אישית לא מוצגים במסוף.

  • מגבלות על קישורים: הקישורים מקובצים לפי אירועים, עד 100 קישורים לכל אירוע. המספר המקסימלי של קישורים ברמת הטבלה הוא 1,000. במנועים אוטומטיים נתמכים שאוספים שושלת יוחסין ברמת העמודה, אם הודעה מכילה יותר מ-1,500 קישורים ברמת העמודה, המערכת מדלגת על המידע ברמת העמודה.

  • היקף התרשים: ב-Knowledge Catalog מוצג תרשים של שושלת היוחסין לכל הרצת משימה, עם הקלט והפלט של אירועי שושלת היוחסין. היא לא תומכת בתהליכים ברמה נמוכה יותר, כמו שלבי Spark.

מיפוי מאפייני פאסט של OpenLineage

למידע על מיפוי OpenLineage, ראו מיפוי OpenLineage.

ייבוא אירוע OpenLineage

אם עדיין לא הגדרתם את OpenLineage, קראו את המאמר תחילת השימוש.

כדי לייבא אירוע OpenLineage ל-Knowledge Catalog, צריך להפעיל את ה-method‏ ProcessOpenLineageRunEvent של ה-API.

C#‎

C#

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

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

using Google.Cloud.DataCatalog.Lineage.V1;
using Google.Protobuf.WellKnownTypes;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for ProcessOpenLineageRunEvent</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 ProcessOpenLineageRunEventRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        ProcessOpenLineageRunEventRequest request = new ProcessOpenLineageRunEventRequest
        {
            Parent = "",
            OpenLineage = new Struct(),
        };
        // Make the request
        ProcessOpenLineageRunEventResponse response = lineageClient.ProcessOpenLineageRunEvent(request);
    }
}

המשך

Go

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

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


//go:build examples

package main

import (
	"context"

	lineage "cloud.google.com/go/datacatalog/lineage/apiv1"
	lineagepb "cloud.google.com/go/datacatalog/lineage/apiv1/lineagepb"
)

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 := lineage.NewClient(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	defer c.Close()

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

Java

Java

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

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

import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.ProcessOpenLineageRunEventRequest;
import com.google.cloud.datacatalog.lineage.v1.ProcessOpenLineageRunEventResponse;
import com.google.protobuf.Struct;

public class SyncProcessOpenLineageRunEvent {

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

  public static void syncProcessOpenLineageRunEvent() 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()) {
      ProcessOpenLineageRunEventRequest request =
          ProcessOpenLineageRunEventRequest.newBuilder()
              .setParent("parent-995424086")
              .setOpenLineage(Struct.newBuilder().build())
              .setRequestId("requestId693933066")
              .build();
      ProcessOpenLineageRunEventResponse response =
          lineageClient.processOpenLineageRunEvent(request);
    }
  }
}

Python

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_process_open_lineage_run_event():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.ProcessOpenLineageRunEventRequest(
        parent="parent_value",
    )

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

    # Handle the response
    print(response)

Ruby

Ruby

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

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

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

##
# Snippet for the process_open_lineage_run_event 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#process_open_lineage_run_event.
#
def process_open_lineage_run_event
  # 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::ProcessOpenLineageRunEventRequest.new

  # Call the process_open_lineage_run_event method.
  result = client.process_open_lineage_run_event request

  # The returned object is of type Google::Cloud::DataCatalog::Lineage::V1::ProcessOpenLineageRunEventResponse.
  p result
end

REST

כדי לייבא אירוע OpenLineage, משתמשים בשיטה processOpenLineageRunEvent.

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

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
  • ‫LOCATION_ID: המיקום של Google Cloud , למשל us-central1.

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

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

תוכן בקשת JSON: 

{
  "eventTime": "2023-04-04T13:21:16.098Z",
  "eventType": "COMPLETE",
  "inputs": [
    {
      "name": "somename",
      "namespace": "customnamespace"
    }
  ],
  "job": {
    "name": "somename",
    "namespace": "customnamespace"
  },
  "outputs": [
    {
      "name": "somename",
      "namespace": "customnamespace"
    }
  ],
  "producer": "someproducer",
  "run": {
    "runId": "somerunid"
  },
  "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/$defs/RunEvent"
}

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

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

{
  "process": "projects/my-project/locations/us-central1/processes/my-process",
  "run": "projects/my-project/locations/us-central1/processes/my-process/runs/my-run",
  "lineageEvents": [
    "projects/my-project/locations/us-central1/processes/my-process/runs/my-run/lineageEvents/my-lineage-event"
  ]
}

כלים לשליחת הודעות OpenLineage

כדי לפשט את שליחת האירועים ל-Data Lineage API, אתם יכולים להשתמש בכלים ובספריות שונים:

  • ספריות הלקוח של Google Cloud ל-Data Lineage: ‏Google מספקת ספריות לקוח כדי ליצור אינטראקציה עם Data Lineage API באופן פרוגרמטי. הוראות להתקנת ספריות לקוח
  • ‫ספריית Google Cloud Java Producer: Google מספקת ספריית Java בקוד פתוח שעוזרת ליצור ולשלוח אירועי OpenLineage ל-Data Lineage API. למידע נוסף, קראו את הפוסט בבלוג ספריית Java של Producer ל-Data Lineage היא עכשיו בקוד פתוח. הספרייה זמינה ב-GitHub וב-Maven.
  • ‫OpenLineage GCP Transport: ליוצרי OpenLineage שמבוססים על Java, יש GcpLineage Transport ייעודי. הוא מפשט את השילוב עם Data Lineage API, על ידי צמצום הקוד שנדרש לשליחת אירועים ל-Data Lineage API. אפשר להגדיר את GcpLineageTransport ככיור האירועים של כל יצרן OpenLineage קיים, כמו Airflow, ‏ Spark ו-Flink. למידע נוסף ולדוגמאות, ראו GcpLineage.

ניתוח מידע מ-OpenLineage

כדי לנתח את אירועי OpenLineage שיובאו, קראו את המאמר הצגת תרשימי שושלת נתונים בממשק המשתמש של Knowledge Catalog.

נתוני פאסטים של OpenLineage שנשמרו

‫Data Lineage API לא מאחסן את כל נתוני ההיבטים מההודעות של OpenLineage. ‫Data Lineage API מאחסן את שדות ההיבטים הבאים:

  • spark_version
    • openlineage-spark-version
    • spark-version
  • כל spark.logicalPlan.*
  • ‫environment-properties (היבט מותאם אישית Google Cloud של שושלת נתונים)
    • origin.sourcetype וגם origin.name
    • spark.app.id
    • spark.app.name
    • spark.batch.id
    • spark.batch.uuid
    • spark.cluster.name
    • spark.cluster.region
    • spark.job.id
    • spark.job.uuid
    • spark.project.id
    • spark.query.node.name
    • spark.session.id
    • spark.session.uuid

ב-Data Lineage API נשמר המידע הבא:

  • eventTime
  • run.runId
  • job.namespace
  • job.name

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