שילוב עם OpenLineage

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

סקירה כללית

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

באמצעות 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 של ProcessOpenLineageRunEvent API בארכיטקטורת REST ולמפות את המאפיינים של OpenLineage למאפיינים של Data Lineage API.

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

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

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

  • תכונות שלא נתמכות: ה-API של מעקב מקורות נתונים לא תומך בתכונות הבאות:

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

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

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

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

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

מיפוי מאפיינים של היבט OpenLineage

מידע על מיפוי OpenLineage זמין במאמר מיפוי OpenLineage.

ייבוא אירוע OpenLineage

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

כדי לייבא אירוע OpenLineage אל Knowledge Catalog, מפעילים את שיטת ה-API‏ ProcessOpenLineageRunEvent.

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 Library:‏ Google מספקת ספריית Java בקוד פתוח שעוזרת ליצור ולשלוח אירועי OpenLineage אל Data Lineage API. מידע נוסף זמין בפוסט בבלוג Producer java library for Data Lineage is now open source. הספרייה זמינה ב-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

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