פיתוח מחבר מותאם אישית לייבוא מטא-נתונים

מחברים בהתאמה אישית מאפשרים לחלץ מטא-נתונים ממקורות נתונים של צד שלישי, כמו MySQL,‏ SQL Server ו-Oracle. אחר כך אתם יכולים לייבא את המטא-נתונים האלה ל-Knowledge Catalog (לשעבר Dataplex Universal Catalog) באמצעות צינור עיבוד נתונים מנוהל לקישוריות. פיתוח מחבר בהתאמה אישית עוזר לכם לקטלג, לחפש ולנהל נכסים ממקורות נתונים שאין להם מחבר מובנה.

אתם יכולים להשתמש בתבנית העזר של PySpark ובמחבר לדוגמה של Python ל-Oracle Database Express Edition (XE) שבדף הזה כנקודת התחלה ליצירת מחבר משלכם.

איך המחברים פועלים

המחבר מחלץ מטא-נתונים ממקור נתונים של צד שלישי, ממיר את המטא-נתונים לפורמט של Knowledge Catalog ImportItem ויוצר קובצי ייבוא של מטא-נתונים שאפשר לייבא באמצעות Knowledge Catalog.

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

צינור עיבוד הנתונים של הקישוריות המנוהלת מריץ את המחבר באמצעות משימה באצווה של Managed Service for Apache Spark. ‫Managed Service for Apache Spark מספק סביבת הרצה של Spark ללא שרת (serverless). אפשר ליצור מחבר שלא משתמש ב-Spark, אבל מומלץ להשתמש ב-Spark כי הוא יכול לשפר את הביצועים של המחבר. אתם יכולים לפתח מחברים באמצעות Python,‏ Java,‏ Scala או R.

דרישות לגבי מחברים

הדרישות של המחבר:

  • המחבר חייב להיות קובץ אימג' של Artifact Registry שאפשר להריץ ב-Managed Service for Apache Spark.
  • המחבר צריך ליצור קובצי מטא-נתונים בפורמט שאפשר לייבא באמצעות משימת ייבוא מטא-נתונים של Knowledge Catalog (ה-method של metadataJobs.create API). דרישות מפורטות מופיעות במאמר קובץ ייבוא מטא-נתונים.

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

    ארגומנט בשורת הפקודה הערך שהצינור מספק
    target_project_id PROJECT_ID
    target_location_id REGION
    target_entry_group_id ENTRY_GROUP_ID
    output_bucket CLOUD_STORAGE_BUCKET_ID
    output_folder FOLDER_ID

    המחבר משתמש בארגומנטים האלה כדי ליצור מטא-נתונים בקבוצת רשומות היעד projects/PROJECT_ID/locations/REGION/entryGroups/ENTRY_GROUP_ID, ולכתוב לקטגוריה של Cloud Storage gs://CLOUD_STORAGE_BUCKET_ID/FOLDER_ID. כל הפעלה של צינור עיבוד הנתונים יוצרת תיקייה חדשה FOLDER_ID בקטגוריה CLOUD_STORAGE_BUCKET_ID. המחבר צריך לכתוב קובצי ייבוא של מטא-נתונים בתיקייה הזו.

תבניות צינור עיבוד הנתונים תומכות במחברי PySpark. התבניות מניחות שהדרייבר (mainPythonFileUri) הוא קובץ מקומי בתמונת המחבר שנקראת main.py. אפשר לשנות את תבניות הצינורות (pipeline) לתרחישים אחרים, כמו מחבר Spark, מזהה URI אחר של מנהל התקן או אפשרויות אחרות.

כך משתמשים ב-PySpark כדי ליצור פריט ייבוא בקובץ ייבוא המטא-נתונים.

"""PySpark schemas for the data."""
entry_source_schema = StructType([
      StructField("display_name", StringType()),
      StructField("source", StringType())])

aspect_schema = MapType(StringType(),
                        StructType([
                            StructField("aspect_type", StringType()),
                            StructField("data", StructType([
                            ]))
                          ])
                        )

entry_schema = StructType([
  StructField("name", StringType()),
  StructField("entry_type", StringType()),
  StructField("fully_qualified_name", StringType()),
  StructField("parent_entry", StringType()),
  StructField("entry_source", entry_source_schema),
  StructField("aspects", aspect_schema)
])

import_item_schema = StructType([
  StructField("entry", entry_schema),
  StructField("aspect_keys", ArrayType(StringType())),
  StructField("update_mask", ArrayType(StringType()))
])

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

בדוגמה שבדף הזה נעשה שימוש במחבר Python ל-Oracle Database Express Edition (XE), והיא מבוססת על ההנחה שאתם מכירים את Python ואת PySpark.

בודקים את הפרטים הבאים:

מבצעים את הפעולות הבאות: יוצרים את כל המשאבים באותו מיקום ב- Google Cloud.

  1. יוצרים או בוחרים פרויקט ב- Google Cloud .

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

    • בחירת פרויקט: כדי לבחור פרויקט לא נדרש תפקיד IAM ספציפי – אתם יכולים לבחור כל פרויקט שקיבלתם בו תפקיד.
    • יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד "יצירת פרויקטים" (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים?

    • יוצרים פרויקט ב- Google Cloud :

      gcloud projects create PROJECT_ID

      מחליפים את PROJECT_ID בשם של הפרויקט ב- Google Cloud שאתם יוצרים.

    • בוחרים את הפרויקט שיצרתם ב- Google Cloud :

      gcloud config set project PROJECT_ID

      מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .

  2. מוודאים שהחיוב מופעל בפרויקט ב- Google Cloud .

  3. מפעילים את ממשקי ה-API של Dataplex,‏ Dataproc,‏ Workflows ו-Artifact Registry, אם הם עדיין לא מופעלים:

    התפקידים שנדרשים כדי להפעיל ממשקי API

    כדי להפעיל ממשקי API, צריך את ההרשאה serviceusage.services.enable. אם אתם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד "בעלים" (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד "אדמין בממשק 'שימוש בשירות'" (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים 

    gcloud services enable dataplex.googleapis.com dataproc.googleapis.com workflows.googleapis.com artifactregistry.googleapis.com

  4. התקינו את ה-CLI של Google Cloud.

  5. אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

  6. כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה: 

    gcloud init

  7. נותנים תפקידים לחשבון המשתמש. מריצים את הפקודה הבאה לכל אחד מהתפקידים הבאים ב-IAM: roles/resourcemanager.projectCreator, roles/billing.projectManager, roles/serviceusage.admin, roles/iam.serviceAccountCreator, roles/iam.securityAdmin, roles/storage.admin, roles/artifactregistry.writer, roles/dataplex.entryGroupOwner, roles/dataplex.entryOwner, roles/dataplex.aspectTypeOwner 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

    מחליפים את מה שכתוב בשדות הבאים:

    • ‫PROJECT_ID: מזהה הפרויקט.
    • ‫USER_IDENTIFIER: המזהה של חשבון המשתמש . לדוגמה, myemail@example.com.
    • ‫ROLE: התפקיד ב-IAM שאתם נותנים לחשבון המשתמש.

  8. מגדירים את האימות:

    1. מוודאים שיש לכם את התפקיד "יצירת חשבונות שירות" (roles/iam.serviceAccountCreator) ואת התפקיד "אדמין של פרויקט" (roles/resourcemanager.projectIamAdmin) ב-IAM. איך מקצים תפקידים

    2. יוצרים את חשבון השירות: 

      gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

      מחליפים את הערך SERVICE_ACCOUNT_NAME בשם שרוצים לתת לחשבון השירות.

    3. מקצים לחשבון השירות את התפקיד roles/owner ב-IAM: 

      gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com" --role=roles/owner

      מחליפים את מה שכתוב בשדות הבאים:

      • SERVICE_ACCOUNT_NAME: השם של חשבון השירות
      • PROJECT_ID: מזהה הפרויקט שבו יצרתם את חשבון השירות

  9. יוצרים קטגוריה של Cloud Storage כדי לאחסן את קובצי הייבוא של המטא-נתונים.

  10. יוצרים את משאבי המטא-נתונים הבאים באותו פרויקט.

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

    1. יוצרים קבוצת רשומות.

    2. יוצרים סוגי היבטים בהתאמה אישית לרשומות שרוצים לייבא. משתמשים במוסכמה למתן שמות SOURCE-ENTITY_TO_IMPORT.

      לדוגמה, למסד נתונים של Oracle, יוצרים סוג היבט בשם oracle-database.

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

    3. יוצרים סוגי רשומות בהתאמה אישית למשאבים שרוצים לייבא, ומקצים להם את סוגי ההיבטים הרלוונטיים. משתמשים במוסכמה למתן שמות SOURCE-ENTITY_TO_IMPORT.

      לדוגמה, למסד נתונים של Oracle, יוצרים סוג רשומה בשם oracle-database. מקשרים אותו לסוג ההיבט שנקרא oracle-database.

  11. מוודאים שאפשר לגשת למקור של הצד השלישי מהפרויקט ב- Google Cloud . מידע נוסף מופיע במאמר הגדרת הרשת של Managed Service for Apache Spark.

יצירת מחבר בסיסי ב-Python

המחבר הבסיסי לדוגמה ב-Python יוצר רשומות ברמה העליונה למקור נתונים של Oracle באמצעות המחלקות של ספריית הלקוח של Knowledge Catalog. לאחר מכן, אתם מספקים את הערכים לשדות הקלט.

המחבר יוצר קובץ ייבוא של מטא-נתונים עם הרשומות הבאות:

  • רשומה מסוג instance, עם סוג הרשומה projects/PROJECT_ID/locations/LOCATION/entryTypes/oracle-instance. הרשומה הזו מייצגת מערכת Oracle Database XE.
  • רשומה של database, שמייצגת מסד נתונים במערכת Oracle Database XE.

כדי ליצור מחבר בסיסי ב-Python:

  1. משכפלים את המאגר cloud-dataplex.

  2. מגדירים סביבה מקומית. מומלץ להשתמש בסביבה וירטואלית.

    mkdir venv
python -m venv venv/
source venv/bin/activate

    משתמשים בגרסאות הפעילות או התחזוקה של Python. יש תמיכה בגרסאות Python 3.7 ואילך.

  3. יוצרים פרויקט ב-Python.

  4. דרישות ההתקנה:

    pip install -r requirements.txt

    הדרישות הבאות מותקנות:

    google-cloud-dataplex==2.2.2
google-cloud-storage
google-cloud-secret-manager

  5. מוסיפים קובץ צינור עיבוד נתונים של main.py בתיקיית הבסיס של הפרויקט.

    from src import bootstrap


if __name__ == '__main__':
    bootstrap.run()

    כשפורסים את הקוד ב-Managed Service for Apache Spark, הקובץ main.py משמש כנקודת הכניסה להפעלה. מומלץ לצמצם את כמות המידע שמאוחסן בקובץ main.py. השתמשו בקובץ הזה כדי לקרוא לפונקציות ולמחלקות שמוגדרות בתוך המחבר, כמו המחלקה src/bootstrap.py.

  6. יוצרים תיקייה בשם src כדי לאחסן בה את רוב הלוגיקה של המחבר.

  7. מעדכנים את הקובץ src/cmd_reader.py באמצעות מחלקת Python כדי לקבל ארגומנטים של שורת הפקודה. אפשר להשתמש במודול argeparse כדי לעשות את זה.

    """Command line reader."""
import argparse


def read_args():
    """Reads arguments from the command line."""
    parser = argparse.ArgumentParser()

    # Dataplex arguments
    parser.add_argument("--target_project_id", type=str, required=True,
        help="The name of the target Google Cloud project to import the metadata into.")
    parser.add_argument("--target_location_id", type=str, required=True,
        help="The target Google Cloud location where the metadata will be imported into.")
    parser.add_argument("--target_entry_group_id", type=str, required=True,
        help="The ID of the entry group to import metadata into. "
             "The metadata will be imported into entry group with the following"
             "full resource name: projects/${target_project_id}/"
             "locations/${target_location_id}/entryGroups/${target_entry_group_id}.")

    # Oracle arguments
    parser.add_argument("--host_port", type=str, required=True,
        help="Oracle host and port number separated by the colon (:).")
    parser.add_argument("--user", type=str, required=True, help="Oracle User.")
    parser.add_argument("--password-secret", type=str, required=True,
        help="Secret resource name in the Secret Manager for the Oracle password.")
    parser.add_argument("--database", type=str, required=True,
        help="Source Oracle database.")

    # Google Cloud Storage arguments
    # It is assumed that the bucket is in the same region as the entry group
    parser.add_argument("--output_bucket", type=str, required=True,
        help="The Cloud Storage bucket to write the generated metadata import file.")
    parser.add_argument("--output_folder", type=str, required=True,
        help="A folder in the Cloud Storage bucket, to write the generated metadata import files.")

    return vars(parser.parse_known_args()[0])

    בסביבות ייצור, מומלץ לשמור את הסיסמה ב-Secret Manager.

  8. מעדכנים את הקובץ src/constants.py באמצעות קוד כדי ליצור קבועים.

    """Constants that are used in the different files."""
import enum

SOURCE_TYPE = "oracle"

# Symbols for replacement
FORBIDDEN = "#"
ALLOWED = "!"


class EntryType(enum.Enum):
    """Types of Oracle entries."""
    INSTANCE: str = "projects/{project}/locations/{location}/entryTypes/oracle-instance"
    DATABASE: str = "projects/{project}/locations/{location}/entryTypes/oracle-database"
    DB_SCHEMA: str = "projects/{project}/locations/{location}/entryTypes/oracle-schema"
    TABLE: str = "projects/{project}/locations/{location}/entryTypes/oracle-table"
    VIEW: str = "projects/{project}/locations/{location}/entryTypes/oracle-view"

  9. מעדכנים את הקובץ src/name_builder.py בשיטות ליצירת משאבי המטא-נתונים שרוצים שהמחבר ייצור בשביל המשאבים ב-Oracle. השתמשו במוסכמות שמתוארות בקטע משאבי מטא-נתונים לדוגמה למקור Oracle במסמך הזה.

    """Builds Dataplex hierarchy identifiers."""
from typing import Dict
from src.constants import EntryType, SOURCE_TYPE


# Oracle cluster users start with C## prefix, but Dataplex doesn't accept #.
# In that case in names it is changed to C!!, and escaped with backticks in FQNs
FORBIDDEN_SYMBOL = "#"
ALLOWED_SYMBOL = "!"


def create_fqn(config: Dict[str, str], entry_type: EntryType,
               schema_name: str = "", table_name: str = ""):
    """Creates a fully qualified name or Dataplex v1 hierarchy name."""
    if FORBIDDEN_SYMBOL in schema_name:
        schema_name = f"`{schema_name}`"

    if entry_type == EntryType.INSTANCE:
        # Requires backticks to escape column
        return f"{SOURCE_TYPE}:`{config['host_port']}`"
    if entry_type == EntryType.DATABASE:
        instance = create_fqn(config, EntryType.INSTANCE)
        return f"{instance}.{config['database']}"
    if entry_type == EntryType.DB_SCHEMA:
        database = create_fqn(config, EntryType.DATABASE)
        return f"{database}.{schema_name}"
    if entry_type in [EntryType.TABLE, EntryType.VIEW]:
        database = create_fqn(config, EntryType.DATABASE)
        return f"{database}.{schema_name}.{table_name}"
    return ""


def create_name(config: Dict[str, str], entry_type: EntryType,
                schema_name: str = "", table_name: str = ""):
    """Creates a Dataplex v2 hierarchy name."""
    if FORBIDDEN_SYMBOL in schema_name:
        schema_name = schema_name.replace(FORBIDDEN_SYMBOL, ALLOWED_SYMBOL)
    if entry_type == EntryType.INSTANCE:
        name_prefix = (
            f"projects/{config['target_project_id']}/"
            f"locations/{config['target_location_id']}/"
            f"entryGroups/{config['target_entry_group_id']}/"
            f"entries/"
        )
        return name_prefix + config["host_port"].replace(":", "@")
    if entry_type == EntryType.DATABASE:
        instance = create_name(config, EntryType.INSTANCE)
        return f"{instance}/databases/{config['database']}"
    if entry_type == EntryType.DB_SCHEMA:
        database = create_name(config, EntryType.DATABASE)
        return f"{database}/database_schemas/{schema_name}"
    if entry_type == EntryType.TABLE:
        db_schema = create_name(config, EntryType.DB_SCHEMA, schema_name)
        return f"{db_schema}/tables/{table_name}"
    if entry_type == EntryType.VIEW:
        db_schema = create_name(config, EntryType.DB_SCHEMA, schema_name)
        return f"{db_schema}/views/{table_name}"
    return ""


def create_parent_name(config: Dict[str, str], entry_type: EntryType,
                       parent_name: str = ""):
    """Generates a Dataplex v2 name of the parent."""
    if entry_type == EntryType.DATABASE:
        return create_name(config, EntryType.INSTANCE)
    if entry_type == EntryType.DB_SCHEMA:
        return create_name(config, EntryType.DATABASE)
    if entry_type == EntryType.TABLE:
        return create_name(config, EntryType.DB_SCHEMA, parent_name)
    return ""


def create_entry_aspect_name(config: Dict[str, str], entry_type: EntryType):
    """Generates an entry aspect name."""
    last_segment = entry_type.value.split("/")[-1]
    return f"{config['target_project_id']}.{config['target_location_id']}.{last_segment}"

    מכיוון שהקובץ name_builder.py משמש גם לקוד הליבה של Python וגם לקוד הליבה של PySpark, מומלץ לכתוב את השיטות כפונקציות טהורות, ולא כחברים במחלקה.

  10. מעדכנים את הקובץ src/top_entry_builder.py באמצעות קוד כדי למלא את הרשומות ברמה העליונה בנתונים.

    """Non-Spark approach for building the entries."""
import dataclasses
import json
from typing import List, Dict

import proto
from google.cloud import dataplex_v1

from src.constants import EntryType
from src import name_builder as nb


@dataclasses.dataclass(slots=True)
class ImportItem:
    """A template class for Import API."""

    entry: dataplex_v1.Entry = dataclasses.field(default_factory=dataplex_v1.Entry)
    aspect_keys: List[str] = dataclasses.field(default_factory=list)
    update_mask: List[str] = dataclasses.field(default_factory=list)


def _dict_factory(data: object):
    """Factory function required for converting Entry dataclass to dict."""

    def convert(obj: object):
        if isinstance(obj, proto.Message):
            return proto.Message.to_dict(obj)
        return obj

    return dict((k, convert(v)) for k, v in data)


def _create_entry(config: Dict[str, str], entry_type: EntryType):
    """Creates an entry based on a Dataplex library."""
    entry = dataplex_v1.Entry()
    entry.name = nb.create_name(config, entry_type)
    entry.entry_type = entry_type.value.format(
        project=config["target_project_id"], location=config["target_location_id"]
    )
    entry.fully_qualified_name = nb.create_fqn(config, entry_type)
    entry.parent_entry = nb.create_parent_name(config, entry_type)

    aspect_key = nb.create_entry_aspect_name(config, entry_type)

    # Add mandatory aspect
    entry_aspect = dataplex_v1.Aspect()
    entry_aspect.aspect_type = aspect_key
    entry_aspect.data = {}
    entry.aspects[aspect_key] = entry_aspect

    return entry


def _entry_to_import_item(entry: dataplex_v1.Entry):
    """Packs entry to import item, accepted by the API,"""
    import_item = ImportItem()
    import_item.entry = entry
    import_item.aspect_keys = list(entry.aspects.keys())
    import_item.update_mask = "aspects"

    return import_item


def create(config, entry_type: EntryType):
    """Creates an entry, packs it to Import Item and converts to json."""
    import_item = _entry_to_import_item(_create_entry(config, entry_type))
    return json.dumps(dataclasses.asdict(import_item, dict_factory=_dict_factory))

  11. מעדכנים את הקובץ src/bootstrap.py באמצעות קוד כדי ליצור את קובץ ייבוא המטא-נתונים ולהפעיל את המחבר.

    """The entrypoint of a pipeline."""
from typing import Dict

from src.constants import EntryType
from src import cmd_reader
from src import secret_manager
from src import entry_builder
from src import gcs_uploader
from src import top_entry_builder
from src.oracle_connector import OracleConnector


FILENAME = "output.jsonl"


def write_jsonl(output_file, json_strings):
    """Writes a list of string to the file in JSONL format."""

    # For simplicity, dataset is written into the one file. But it is not
    # mandatory, and the order doesn't matter for Import API.
    # The PySpark itself could dump entries into many smaller JSONL files.
    # Due to performance, it's recommended to dump to many smaller files.
    for string in json_strings:
        output_file.write(string + "\n")


def process_dataset(
    connector: OracleConnector,
    config: Dict[str, str],
    schema_name: str,
    entry_type: EntryType,
):
    """Builds dataset and converts it to jsonl."""
    df_raw = connector.get_dataset(schema_name, entry_type)
    df = entry_builder.build_dataset(config, df_raw, schema_name, entry_type)
    return df.toJSON().collect()


def run():
    """Runs a pipeline."""
    config = cmd_reader.read_args()
    config["password"] = secret_manager.get_password(config["password_secret"])
    connector = OracleConnector(config)

    with open(FILENAME, "w", encoding="utf-8") as file:
        # Write top entries that don't require connection to the database
        file.writelines(top_entry_builder.create(config, EntryType.INSTANCE))
        file.writelines("\n")
        file.writelines(top_entry_builder.create(config, EntryType.DATABASE))

        # Get schemas, write them and collect to the list
        df_raw_schemas = connector.get_db_schemas()
        schemas = [schema.USERNAME for schema in df_raw_schemas.select("USERNAME").collect()]
        schemas_json = entry_builder.build_schemas(config, df_raw_schemas).toJSON().collect()

        write_jsonl(file, schemas_json)

        # Ingest tables and views for every schema in a list
        for schema in schemas:
            print(f"Processing tables for {schema}")
            tables_json = process_dataset(connector, config, schema, EntryType.TABLE)
            write_jsonl(file, tables_json)
            print(f"Processing views for {schema}")
            views_json = process_dataset(connector, config, schema, EntryType.VIEW)
            write_jsonl(file, views_json)

    gcs_uploader.upload(config, FILENAME)

  12. מריצים את הקוד באופן מקומי.

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

  13. אופציונלי: הרחבת הדוגמה הקודמת כדי להשתמש במחלקות של ספריית הלקוח של Knowledge Catalog ליצירת פריטי ייבוא לטבלאות, לסכימות ולתצוגות. אתם יכולים גם להריץ את הדוגמה ב-Python ב-Managed Service for Apache Spark.

    אנחנו ממליצים ליצור מחבר שמשתמש ב-Spark (ופועל ב-Managed Service for Apache Spark), כי הוא יכול לשפר את הביצועים של המחבר.

יצירת מחבר PySpark

הדוגמה הזו מבוססת על PySpark DataFrame API. אתם יכולים להתקין את PySpark SQL ולהריץ אותו באופן מקומי לפני שתריצו אותו ב-Managed Service for Apache Spark. אם אתם מתקינים ומריצים את PySpark באופן מקומי, אתם צריכים להתקין את ספריית PySpark באמצעות pip, אבל לא צריך להתקין אשכול Spark מקומי.

מסיבות שקשורות לביצועים, בדוגמה הזו לא נעשה שימוש במחלקות מוגדרות מראש מספריית PySpark. במקום זאת, בדוגמה נוצרים DataFrames, הם מומרים לרשומות JSON, ואז הפלט נכתב לקובץ ייבוא של מטא-נתונים בפורמט JSON Lines שאפשר לייבא ל-Knowledge Catalog.

כדי ליצור מחבר באמצעות PySpark:

  1. משכפלים את המאגר cloud-dataplex.

  2. מתקינים את PySpark:

    pip install pyspark

  3. דרישות ההתקנה:

    pip install -r requirements.txt

    הדרישות הבאות מותקנות:

    google-cloud-dataplex==2.2.2
google-cloud-storage
google-cloud-secret-manager

  4. מעדכנים את הקובץ oracle_connector.py באמצעות קוד כדי לקרוא נתונים ממקור נתונים של Oracle ולהחזיר DataFrames.

    """Reads Oracle using PySpark."""
from typing import Dict
from pyspark.sql import SparkSession, DataFrame

from src.constants import EntryType


SPARK_JAR_PATH = "/opt/spark/jars/ojdbc11.jar"


class OracleConnector:
    """Reads data from Oracle and returns Spark Dataframes."""

    def __init__(self, config: Dict[str, str]):
        # PySpark entrypoint
        self._spark = SparkSession.builder.appName("OracleIngestor") \
            .config("spark.jars", SPARK_JAR_PATH) \
            .getOrCreate()

        self._config = config
        self._url = f"jdbc:oracle:thin:@{config['host_port']}:{config['database']}"

    def _execute(self, query: str) -> DataFrame:
        """A generic method to execute any query."""
        return self._spark.read.format("jdbc") \
            .option("driver", "oracle.jdbc.OracleDriver") \
            .option("url", self._url) \
            .option("query", query) \
            .option("user", self._config["user"]) \
            .option("password", self._config["password"]) \
            .load()

    def get_db_schemas(self) -> DataFrame:
        """In Oracle, schemas are usernames."""
        query = "SELECT username FROM dba_users"
        return self._execute(query)

    def _get_columns(self, schema_name: str, object_type: str) -> str:
        """Gets a list of columns in tables or views in a batch."""
        # Every line here is a column that belongs to the table or to the view.
        # This SQL gets data from ALL the tables in a given schema.
        return (f"SELECT col.TABLE_NAME, col.COLUMN_NAME, "
                f"col.DATA_TYPE, col.NULLABLE "
                f"FROM all_tab_columns col "
                f"INNER JOIN DBA_OBJECTS tab "
                f"ON tab.OBJECT_NAME = col.TABLE_NAME "
                f"WHERE tab.OWNER = '{schema_name}' "
                f"AND tab.OBJECT_TYPE = '{object_type}'")

    def get_dataset(self, schema_name: str, entry_type: EntryType):
        """Gets data for a table or a view."""
        # Dataset means that these entities can contain end user data.
        short_type = entry_type.name  # table or view, or the title of enum value
        query = self._get_columns(schema_name, short_type)
        return self._execute(query)

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

    • סכימות של מסדי נתונים
    • טבלאות ששייכות לסכימות האלה
    • העמודות ששייכות לטבלאות האלה, כולל שם העמודה, סוג הנתונים בעמודה, והאם העמודה יכולה להיות ריקה או שהיא נדרשת

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

    שימו לב לנקודות הבאות:

    • ב-Oracle, סכימה של מסד נתונים נמצאת בבעלות של משתמש במסד הנתונים, והשם שלה זהה לשם של המשתמש.
    • אובייקטים של סכימה הם מבנים לוגיים שנוצרים על ידי משתמשים. אובייקטים כמו טבלאות או אינדקסים יכולים להכיל נתונים, ואובייקטים כמו תצוגות או מילים נרדפות מכילים רק הגדרה.
    • הקובץ ojdbc11.jar מכיל את מנהל ההתקן של Oracle JDBC.

  5. מעדכנים את הקובץ src/entry_builder.py בשיטות משותפות להחלת טרנספורמציות של Spark.

    """Creates entries with PySpark."""
import pyspark.sql.functions as F
from pyspark.sql.types import StringType

from src.constants import EntryType, SOURCE_TYPE
from src import name_builder as nb


@F.udf(returnType=StringType())
def choose_metadata_type_udf(data_type: str):
    """Choose the metadata type based on Oracle native type."""
    if data_type.startswith("NUMBER") or data_type in ["FLOAT", "LONG"]:
        return "NUMBER"
    if data_type.startswith("VARCHAR") or data_type.startswith("NVARCHAR2"):
        return "STRING"
    if data_type == "DATE":
        return "DATETIME"
    return "OTHER"


def create_entry_source(column):
    """Create Entry Source segment."""
    return F.named_struct(F.lit("display_name"),
                          column,
                          F.lit("system"),
                          F.lit(SOURCE_TYPE))


def create_entry_aspect(entry_aspect_name):
    """Create aspect with general information (usually it is empty)."""
    return F.create_map(
        F.lit(entry_aspect_name),
        F.named_struct(
            F.lit("aspect_type"),
            F.lit(entry_aspect_name),
            F.lit("data"),
            F.create_map()
            )
        )


def convert_to_import_items(df, aspect_keys):
    """Convert entries to import items."""
    entry_columns = ["name", "fully_qualified_name", "parent_entry",
                     "entry_source", "aspects", "entry_type"]

    # Puts entry to "entry" key, a list of keys from aspects in "aspects_keys"
    # and "aspects" string in "update_mask"
    return df.withColumn("entry", F.struct(entry_columns)) \
      .withColumn("aspect_keys", F.array([F.lit(key) for key in aspect_keys])) \
      .withColumn("update_mask", F.array(F.lit("aspects"))) \
      .drop(*entry_columns)


def build_schemas(config, df_raw_schemas):
    """Create a dataframe with database schemas from the list of usernames.
    Args:
        df_raw_schemas - a dataframe with only one column called USERNAME
    Returns:
        A dataframe with Dataplex-readable schemas.
    """
    entry_type = EntryType.DB_SCHEMA
    entry_aspect_name = nb.create_entry_aspect_name(config, entry_type)

    # For schema, parent name is the name of the database
    parent_name =  nb.create_parent_name(config, entry_type)

    # Create user-defined function.
    create_name_udf = F.udf(lambda x: nb.create_name(config, entry_type, x),
                            StringType())
    create_fqn_udf = F.udf(lambda x: nb.create_fqn(config, entry_type, x),
                           StringType())

    # Fills the missed project and location into the entry type string
    full_entry_type = entry_type.value.format(
        project=config["target_project_id"],
        location=config["target_location_id"])

    # Converts a list of schema names to the Dataplex-compatible form
    column = F.col("USERNAME")
    df = df_raw_schemas.withColumn("name", create_name_udf(column)) \
      .withColumn("fully_qualified_name", create_fqn_udf(column)) \
      .withColumn("parent_entry", F.lit(parent_name)) \
      .withColumn("entry_type", F.lit(full_entry_type)) \
      .withColumn("entry_source", create_entry_source(column)) \
      .withColumn("aspects", create_entry_aspect(entry_aspect_name)) \
    .drop(column)

    df = convert_to_import_items(df, [entry_aspect_name])
    return df


def build_dataset(config, df_raw, db_schema, entry_type):
    """Build table entries from a flat list of columns.
    Args:
        df_raw - a plain dataframe with TABLE_NAME, COLUMN_NAME, DATA_TYPE,
                 and NULLABLE columns
        db_schema - parent database schema
        entry_type - entry type: table or view
    Returns:
        A dataframe with Dataplex-readable data of tables of views.
    """
    schema_key = "dataplex-types.global.schema"

    # The transformation below does the following
    # 1. Alters NULLABLE content from Y/N to NULLABLE/REQUIRED
    # 2. Renames NULLABLE to mode
    # 3. Renames DATA_TYPE to dataType
    # 4. Creates metadataType column based on dataType column
    # 5. Renames COLUMN_NAME to name
    df = df_raw \
      .withColumn("mode", F.when(F.col("NULLABLE") == 'Y', "NULLABLE").otherwise("REQUIRED")) \
      .drop("NULLABLE") \
      .withColumnRenamed("DATA_TYPE", "dataType") \
      .withColumn("metadataType", choose_metadata_type_udf("dataType")) \
      .withColumnRenamed("COLUMN_NAME", "name")

    # The transformation below aggregate fields, denormalizing the table
    # TABLE_NAME becomes top-level filed, and the rest is put into
    # the array type called "fields"
    aspect_columns = ["name", "mode", "dataType", "metadataType"]
    df = df.withColumn("columns", F.struct(aspect_columns))\
      .groupby('TABLE_NAME') \
      .agg(F.collect_list("columns").alias("fields"))

    # Create nested structured called aspects.
    # Fields are becoming a part of a `schema` struct
    # There is also an entry_aspect that is repeats entry_type as aspect_type
    entry_aspect_name = nb.create_entry_aspect_name(config, entry_type)
    df = df.withColumn("schema",
                       F.create_map(F.lit(schema_key),
                                    F.named_struct(
                                        F.lit("aspect_type"),
                                        F.lit(schema_key),
                                        F.lit("data"),
                                        F.create_map(F.lit("fields"),
                                                     F.col("fields")))
                                    )
                       )\
      .withColumn("entry_aspect", create_entry_aspect(entry_aspect_name)) \
    .drop("fields")

    # Merge separate aspect columns into the one map called 'aspects'
    df = df.select(F.col("TABLE_NAME"),
                   F.map_concat("schema", "entry_aspect").alias("aspects"))

    # Define user-defined functions to fill the general information
    # and hierarchy names
    create_name_udf = F.udf(lambda x: nb.create_name(config, entry_type,
                                                     db_schema, x),
                            StringType())

    create_fqn_udf = F.udf(lambda x: nb.create_fqn(config, entry_type,
                                                   db_schema, x), StringType())

    parent_name = nb.create_parent_name(config, entry_type, db_schema)
    full_entry_type = entry_type.value.format(
        project=config["target_project_id"],
        location=config["target_location_id"])

    # Fill the top-level fields
    column = F.col("TABLE_NAME")
    df = df.withColumn("name", create_name_udf(column)) \
      .withColumn("fully_qualified_name", create_fqn_udf(column)) \
      .withColumn("entry_type", F.lit(full_entry_type)) \
      .withColumn("parent_entry", F.lit(parent_name)) \
      .withColumn("entry_source", create_entry_source(column)) \
    .drop(column)

    df = convert_to_import_items(df, [schema_key, entry_aspect_name])
    return df

    שימו לב לנקודות הבאות:

    • השיטות יוצרות את משאבי המטא-נתונים שהמחבר יוצר בשביל המשאבים שלכם ב-Oracle. השתמשו במוסכמות שמתוארות בקטע משאבי מטא-נתונים לדוגמה למקור Oracle במסמך הזה.
    • השיטה convert_to_import_items רלוונטית לסכימות, לטבלאות ולתצוגות. מוודאים שהפלט של המחבר הוא פריט ייבוא אחד או יותר שאפשר לעבד באמצעות השיטה metadataJobs.create, ולא רשומות נפרדות.
    • גם בתצוגה, העמודה נקראת TABLE_NAME.

  6. מעדכנים את הקובץ bootstrap.py באמצעות קוד כדי ליצור את קובץ ייבוא המטא-נתונים ולהפעיל את המחבר.

    """The entrypoint of a pipeline."""
from typing import Dict

from src.constants import EntryType
from src import cmd_reader
from src import secret_manager
from src import entry_builder
from src import gcs_uploader
from src import top_entry_builder
from src.oracle_connector import OracleConnector


FILENAME = "output.jsonl"


def write_jsonl(output_file, json_strings):
    """Writes a list of string to the file in JSONL format."""

    # For simplicity, dataset is written into the one file. But it is not
    # mandatory, and the order doesn't matter for Import API.
    # The PySpark itself could dump entries into many smaller JSONL files.
    # Due to performance, it's recommended to dump to many smaller files.
    for string in json_strings:
        output_file.write(string + "\n")


def process_dataset(
    connector: OracleConnector,
    config: Dict[str, str],
    schema_name: str,
    entry_type: EntryType,
):
    """Builds dataset and converts it to jsonl."""
    df_raw = connector.get_dataset(schema_name, entry_type)
    df = entry_builder.build_dataset(config, df_raw, schema_name, entry_type)
    return df.toJSON().collect()


def run():
    """Runs a pipeline."""
    config = cmd_reader.read_args()
    config["password"] = secret_manager.get_password(config["password_secret"])
    connector = OracleConnector(config)

    with open(FILENAME, "w", encoding="utf-8") as file:
        # Write top entries that don't require connection to the database
        file.writelines(top_entry_builder.create(config, EntryType.INSTANCE))
        file.writelines("\n")
        file.writelines(top_entry_builder.create(config, EntryType.DATABASE))

        # Get schemas, write them and collect to the list
        df_raw_schemas = connector.get_db_schemas()
        schemas = [schema.USERNAME for schema in df_raw_schemas.select("USERNAME").collect()]
        schemas_json = entry_builder.build_schemas(config, df_raw_schemas).toJSON().collect()

        write_jsonl(file, schemas_json)

        # Ingest tables and views for every schema in a list
        for schema in schemas:
            print(f"Processing tables for {schema}")
            tables_json = process_dataset(connector, config, schema, EntryType.TABLE)
            write_jsonl(file, tables_json)
            print(f"Processing views for {schema}")
            views_json = process_dataset(connector, config, schema, EntryType.VIEW)
            write_jsonl(file, views_json)

    gcs_uploader.upload(config, FILENAME)

    בדוגמה הזו, קובץ הייבוא של המטא-נתונים נשמר כקובץ JSON Lines יחיד. אפשר להשתמש בכלים של PySpark, כמו המחלקה DataFrameWriter, כדי להפיק קבוצות של קובצי JSON במקביל.

    המחבר יכול לכתוב רשומות לקובץ ייבוא המטא-נתונים בכל סדר.

  7. מעדכנים את הקובץ gcs_uploader.py באמצעות קוד כדי להעלות את קובץ ייבוא המטא-נתונים לקטגוריה של Cloud Storage.

    """Sends files to GCP storage."""
from typing import Dict
from google.cloud import storage


def upload(config: Dict[str, str], filename: str):
    """Uploads a file to GCP bucket."""
    client = storage.Client()
    bucket = client.get_bucket(config["output_bucket"])
    folder = config["output_folder"]

    blob = bucket.blob(f"{folder}/{filename}")
    blob.upload_from_filename(filename)

  8. יוצרים את קובץ האימג' של המחבר.

    אם המחבר מכיל כמה קבצים, או אם אתם רוצים להשתמש בספריות שלא כלולות בקובץ האימג' של Docker שמוגדר כברירת מחדל, אתם צריכים להשתמש במאגר בהתאמה אישית. ב-Managed Service for Apache Spark, עומסי העבודה פועלים בתוך קונטיינרים של Docker. יוצרים קובץ אימג' של Docker בהתאמה אישית של המחבר ומאחסנים את קובץ האימג' ב-Artifact Registry. ‫Managed Service for Apache Spark קורא את קובץ האימג' מ-Artifact Registry.

    1. יוצרים קובץ Docker:

      FROM debian:11-slim

ENV DEBIAN_FRONTEND=noninteractive

RUN apt update && apt install -y procps tini
RUN apt install -y wget

ENV SPARK_EXTRA_JARS_DIR=/opt/spark/jars/
RUN mkdir -p "${SPARK_EXTRA_JARS_DIR}"
COPY ojdbc11.jar "${SPARK_EXTRA_JARS_DIR}"

ENV CONDA_HOME=/opt/miniconda3
ENV PYSPARK_PYTHON=${CONDA_HOME}/bin/python
ENV PATH=${CONDA_HOME}/bin:${PATH}
RUN wget https://repo.anaconda.com/miniconda/Miniconda3-py311_24.9.2-0-Linux-x86_64.sh

RUN bash Miniconda3-py310_23.3.1-0-Linux-x86_64.sh -b -p /opt/miniconda3 \
  && ${CONDA_HOME}/bin/conda config --system --set always_yes True \
  && ${CONDA_HOME}/bin/conda config --system --set auto_update_conda False \
  && ${CONDA_HOME}/bin/conda config --system --prepend channels conda-forge \
  && ${CONDA_HOME}/bin/conda config --system --set channel_priority strict

RUN ${CONDA_HOME}/bin/conda install mamba -n base -c conda-forge \
    && ${CONDA_HOME}/bin/mamba install \
      conda \
      google-cloud-dataproc \
      google-cloud-logging \
      google-cloud-monitoring \
      google-cloud-storage

RUN apt update && apt install -y git
COPY requirements.txt .
RUN python -m pip install -r requirements.txt

ENV PYTHONPATH=/opt/python/packages
RUN mkdir -p "${PYTHONPATH}/src/"
COPY src/ "${PYTHONPATH}/src/"
COPY main.py .

RUN groupadd -g 1099 spark
RUN useradd -u 1099 -g 1099 -d /home/spark -m spark
USER spark

      משתמשים ב-Conda ככלי לניהול חבילות. ‫Managed Service for Apache Spark טוען את pyspark לקונטיינר בזמן הריצה, כך שאין צורך להתקין יחסי תלות של PySpark בקובץ האימג' של הקונטיינר בהתאמה אישית.

    2. יוצרים את קובץ האימג' של הקונטיינר בהתאמה אישית ומעבירים אותו בדחיפה ל-Artifact Registry.

      #!/bin/bash

IMAGE=oracle-pyspark:0.0.1
PROJECT=<PROJECT_ID>


REPO_IMAGE=us-central1-docker.pkg.dev/${PROJECT}/docker-repo/oracle-pyspark

docker build -t "${IMAGE}" .

# Tag and push to GCP container registry
gcloud config set project ${PROJECT}
gcloud auth configure-docker us-central1-docker.pkg.dev
docker tag "${IMAGE}" "${REPO_IMAGE}"
docker push "${REPO_IMAGE}"

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

  9. מריצים את המחבר ב-Managed Service for Apache Spark. כדי לשלוח משימה באצווה של PySpark באמצעות קובץ האימג' של הקונטיינר בהתאמה אישית, מריצים את הפקודה gcloud dataproc batches submit pyspark.

    gcloud dataproc batches submit pyspark main.py --project=PROJECT \
    --region=REGION --batch=BATCH_ID \
    --container-image=CUSTOM_CONTAINER_IMAGE \
    --service-account=SERVICE_ACCOUNT_NAME \
    --jars=PATH_TO_JAR_FILES \
    --properties=PYSPARK_PROPERTIES \
    -- PIPELINE_ARGUMENTS

    שימו לב לנקודות הבאות:

    • קובצי ה-JAR הם מנהלי התקנים ל-Spark. כדי לקרוא מ-Oracle, מ-MySQL או מ-Postgres, צריך לספק ל-Apache Spark חבילה ספציפית. החבילה יכולה להיות ב-Cloud Storage או בתוך הקונטיינר. אם קובץ ה-JAR נמצא בתוך הקונטיינר, הנתיב דומה ל-file:///path/to/file/driver.jar. בדוגמה הזו, הנתיב לקובץ ה-JAR הוא /opt/spark/jars/.
    • ‫PIPELINE_ARGUMENTS הם הארגומנטים בשורת הפקודה של המחבר.

    המחבר מחלץ מטא-נתונים ממסד הנתונים של Oracle, יוצר קובץ לייבוא מטא-נתונים ושומר אותו בקטגוריה של Cloud Storage.

  10. כדי לייבא באופן ידני את המטא-נתונים מקובץ הייבוא אל Knowledge Catalog, צריך להריץ משימת מטא-נתונים. משתמשים בשיטה metadataJobs.create.

    1. בשורת הפקודה, מוסיפים משתני סביבה ויוצרים כינוי לפקודת ה-curl.

      PROJECT_ID=PROJECT
LOCATION_ID=LOCATION
DATAPLEX_API=dataplex.googleapis.com/v1/projects/$PROJECT_ID/locations/$LOCATION_ID
alias gcurl='curl -H "Authorization: Bearer $(gcloud auth print-access-token)" -H "Content-Type: application/json"'

    2. מפעילים את ה-method של ה-API ומעבירים את סוגי הרשומות וסוגי ההיבטים שרוצים לייבא.

      gcurl https://${DATAPLEX_API}/metadataJobs?metadata_job_id="JOB_ID" -d "$(cat <<EOF
{
  "type": "IMPORT",
  "import_spec": {
    "source_storage_uri": "gs://BUCKET/FOLDER/",
    "entry_sync_mode": "FULL",
    "aspect_sync_mode": "INCREMENTAL",
    "scope": {
      "entry_groups": ["projects/PROJECT/locations/LOCATION/entryGroups/ENTRY_GROUP_ID"],
      "entry_types": [
        "projects/PROJECT/locations/LOCATION/entryTypes/oracle-instance",
        "projects/PROJECT/locations/LOCATION/entryTypes/oracle-database",
        "projects/PROJECT/locations/LOCATION/entryTypes/oracle-schema",
        "projects/PROJECT/locations/LOCATION/entryTypes/oracle-table",
        "projects/PROJECT/locations/LOCATION/entryTypes/oracle-view"],

      "aspect_types": [
        "projects/PROJECT/locations/LOCATION/aspectTypes/oracle-instance",
        "projects/dataplex-types/locations/global/aspectTypes/schema",
        "projects/PROJECT/locations/LOCATION/aspectTypes/oracle-database",
        "projects/PROJECT/locations/LOCATION/aspectTypes/oracle-schema",
        "projects/PROJECT/locations/LOCATION/aspectTypes/oracle-table",
        "projects/PROJECT/locations/LOCATION/aspectTypes/oracle-view"],
      },
    },
  }
EOF
)"

      סוג ההיבט schema הוא סוג היבט גלובלי שמוגדר על ידי Knowledge Catalog.

      שימו לב שהפורמט שבו משתמשים לשמות של סוגי ההיבטים כשקוראים ל-method של ה-API שונה מהפורמט שבו משתמשים בקוד של המחבר.

    3. אופציונלי: אפשר להשתמש ב-Cloud Logging כדי לצפות ביומנים של משימת המטא-נתונים. מידע נוסף זמין במאמר מעקב אחרי יומני Knowledge Catalog.

הגדרת תזמור של צינור עיבוד נתונים

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

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

  1. כדי להריץ צינור עיבוד נתונים מנוהל של קישוריות באמצעות המחבר לדוגמה, פועלים לפי השלבים לייבוא מטא-נתונים באמצעות Workflows. כדאי לבצע את הפעולות הבאות:

    • יוצרים את תהליך העבודה באותו Google Cloud מיקום של המחבר.

    • בקובץ ההגדרה של תהליך העבודה, מעדכנים את הפונקציה submit_pyspark_extract_job באמצעות הקוד הבא כדי לחלץ נתונים ממסד הנתונים של Oracle באמצעות המחבר שיצרתם.

      - submit_pyspark_extract_job:
    call: http.post
    args:
      url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
      auth:
        type: OAuth2
        scopes: "https://www.googleapis.com/auth/cloud-platform"
      headers:
        Content-Type: "application/json"
      query:
        batchId: ${WORKFLOW_ID}
      body:
        pysparkBatch:
          mainPythonFileUri: file:///main.py
          jars: file:///opt/spark/jars/ojdbc11.jar
          args:
            - ${"--host_port=" + args.ORACLE_HOST_PORT}
            - ${"--user=" + args.ORACLE_USER}
            - ${"--password=" + args.ORACLE_PASSWORD}
            - ${"--database=" + args.ORACLE_DATABASE}
            - ${"--project=" + args.TARGET_PROJECT_ID}
            - ${"--location=" + args.CLOUD_REGION}
            - ${"--entry_group=" + args.TARGET_ENTRY_GROUP_ID}
            - ${"--bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
            - ${"--folder=" + WORKFLOW_ID}
        runtimeConfig:
          version: "2.0"
          containerImage: "us-central1-docker.pkg.dev/PROJECT/REPOSITORY/oracle-pyspark"
        environmentConfig:
          executionConfig:
              serviceAccount: ${args.SERVICE_ACCOUNT}
    result: RESPONSE_MESSAGE

    • בקובץ ההגדרה של תהליך העבודה, מעדכנים את הפונקציה submit_import_job באמצעות הקוד הבא כדי לייבא את הרשומות. הפונקציה מפעילה את ה-method‏ metadataJobs.create של ה-API כדי להריץ משימת ייבוא של מטא-נתונים.

      - submit_import_job:
    call: http.post
    args:
      url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
      auth:
        type: OAuth2
        scopes: "https://www.googleapis.com/auth/cloud-platform"
      body:
        type: IMPORT
        import_spec:
          source_storage_uri: ${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
          entry_sync_mode: FULL
          aspect_sync_mode: INCREMENTAL
          scope:
            entry_groups:
              - ${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/"+args.TARGET_ENTRY_GROUP_ID}
            entry_types:
              -"projects/PROJECT/locations/LOCATION/entryTypes/oracle-instance"
              -"projects/PROJECT/locations/LOCATION/entryTypes/oracle-database"
              -"projects/PROJECT/locations/LOCATION/entryTypes/oracle-schema"
              -"projects/PROJECT/locations/LOCATION/entryTypes/oracle-table"
              -"projects/PROJECT/locations/LOCATION/entryTypes/oracle-view"
            aspect_types:
              -"projects/PROJECT/locations/LOCATION/aspectTypes/oracle-instance"
              -"projects/dataplex-types/locations/global/aspectTypes/schema"
              -"projects/PROJECT/locations/LOCATION/aspectTypes/oracle-database"
              -"projects/PROJECT/locations/LOCATION/aspectTypes/oracle-schema"
              -"projects/PROJECT/locations/LOCATION/aspectTypes/oracle-table"
              -"projects/PROJECT/locations/LOCATION/aspectTypes/oracle-view"
    result: IMPORT_JOB_RESPONSE

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

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

      {
  "CLOUD_REGION": "us-central1",
  "ORACLE_USER": "system",
  "ORACLE_HOST_PORT": "x.x.x.x:1521",
  "ORACLE_DATABASE": "xe",
  "ADDITIONAL_CONNECTOR_ARGS": [],
}

  2. אופציונלי: אפשר להשתמש ב-Cloud Logging כדי להציג את היומנים של צינור עיבוד הנתונים של הקישוריות המנוהלת. המטען הייעודי (payload) של היומן כולל קישור ליומנים של המשימה באצווה של Managed Service for Apache Spark ושל המשימה לייבוא המטא-נתונים, לפי הצורך. מידע נוסף זמין במאמר הצגת יומני תהליכי עבודה.

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

    1. משתמשים ב-Secret Manager כדי לאחסן את פרטי הכניסה למקור הנתונים של הצד השלישי.
    2. משתמשים ב-PySpark כדי לכתוב את פלט ה-JSON Lines לכמה קובצי ייבוא של מטא-נתונים במקביל.
    3. אפשר להשתמש בקידומת כדי לפצל קבצים גדולים (יותר מ-100MB) לקבצים קטנים יותר.
    4. להוסיף עוד היבטים בהתאמה אישית שמתעדים מטא-נתונים עסקיים וטכניים נוספים מהמקור.

דוגמאות למשאבי מטא-נתונים של מקור Oracle

המחבר לדוגמה מחלץ מטא-נתונים ממסד נתונים של Oracle וממפה את המטא-נתונים למשאבי מטא-נתונים תואמים של Knowledge Catalog.

שיקולים שקשורים להיררכיה

לכל מערכת ב-Knowledge Catalog יש רשומת שורש שהיא רשומת ההורה של המערכת. בדרך כלל, לרשומת הבסיס יש סוג רשומה instance. בטבלה הבאה מוצגת היררכיה לדוגמה של סוגי רשומות וסוגי היבטים של מערכת Oracle. לדוגמה, סוג הרשומה oracle-database מקושר לסוג מאפיין שנקרא גם oracle-database.

מזהה סוג הרשומה תיאור מזהה סוג ההיבט המקושר
oracle-instance השורש של המערכת המיובאת. oracle-instance
oracle-database מסד הנתונים של Oracle. oracle-database
oracle-schema הסכימה של מסד הנתונים. oracle-schema
oracle-table טבלה.

oracle-table

schema
oracle-view תצוגה.

oracle-view

schema

סוג ההיבט schema הוא סוג היבט גלובלי שמוגדר על ידי Knowledge Catalog. הוא מכיל תיאור של השדות בטבלה, בתצוגה או בישות אחרת שיש בה עמודות. סוג ההיבט המותאם אישית oracle-schema מכיל את השם של הסכימה של מסד הנתונים של Oracle.

דוגמה לשדות של פריטים לייבוא

המחבר צריך להשתמש במוסכמות הבאות למשאבי Oracle.

  • שמות שמוגדרים במלואם: שמות שמוגדרים במלואם למשאבים של Oracle משתמשים בתבנית השמות הבאה. תווים אסורים מופיעים עם תווי בריחה (escape) של מירכאות הפוכות.

    משאב תבנית דוגמה
    Instance

    SOURCE:ADDRESS

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

    		 oracle:`localhost:1521` או oracle:`myinstance.com`
    מסד נתונים SOURCE:ADDRESS.DATABASE oracle:`localhost:1521`.xe
    סכימה SOURCE:ADDRESS.DATABASE.SCHEMA oracle:`localhost:1521`.xe.sys
    טבלה SOURCE:ADDRESS.DATABASE.SCHEMA.TABLE_NAME oracle:`localhost:1521`.xe.sys.orders
    הצגה SOURCE:ADDRESS.DATABASE.SCHEMA.VIEW_NAME oracle:`localhost:1521`.xe.sys.orders_view

  • שמות או מזהים של רשומות: רשומות של משאבי Oracle משתמשות בתבנית השמות הבאה. תווים אסורים מוחלפים בתו מותר. המשאבים משתמשים בקידומת projects/PROJECT/locations/LOCATION/entryGroups/ENTRY_GROUP/entries.

    משאב תבנית דוגמה
    Instance HOST_PORT/PREFIX projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521
    מסד נתונים PREFIX/HOST_PORT/databases/DATABASE projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521/databases/xe
    סכימה PREFIX/HOST_PORT/databases/DATABASE/database_schemas/SCHEMA projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521/databases/xe/database_schemas/sys
    טבלה PREFIX/HOST_PORT/databases/DATABASE/database_schemas/SCHEMA/tables/TABLE projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521/databases/xe/database_schemas/sys/tables/orders
    הצגה PREFIX/HOST_PORT/databases/DATABASE/database_schemas/SCHEMA/views/VIEW projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521/databases/xe/database_schemas/sys/views/orders_view

  • רשומות הורה: אם רשומה היא לא רשומת שורש של המערכת, יכול להיות שיהיה לה שדה של רשומת הורה שמתאר את המיקום שלה בהיררכיה. השדה צריך להכיל את השם של רשומת ההורה. מומלץ ליצור את הערך הזה.

    בטבלה הבאה מוצגים רשומות ההורה של משאבי Oracle.

    הערך רשומת הורה
    Instance "" (מחרוזת ריקה)
    מסד נתונים שם המכונה
    סכימה שם מסד הנתונים
    טבלה שם הסכימה
    הצגה שם הסכימה

  • ‫Aspect map: מפת ההיבטים חייבת להכיל לפחות היבט אחד שמתאר את הישות שרוצים לייבא. הנה דוגמה למפת היבטים של טבלת Oracle.

    "example-project.us-central1.oracle-table": {
    "aspect_type": "example-project.us-central1.oracle-table",
    "path": "",
    "data": {}
 },

    אפשר למצוא סוגי היבטים מוגדרים מראש (כמו schema) שמגדירים את מבנה הטבלה או התצוגה בפרויקט dataplex-types, במיקום global.

  • ‫Aspect keys: מפתחות ההיבטים משתמשים בפורמט השמות PROJECT.LOCATION.ASPECT_TYPE. בטבלה הבאה מוצגות דוגמאות למפתחות היבטים של משאבי Oracle.

    הערך מפתח היבט לדוגמה
    Instance example-project.us-central1.oracle-instance
    מסד נתונים example-project.us-central1.oracle-database
    סכימה example-project.us-central1.oracle-schema
    טבלה example-project.us-central1.oracle-table
    הצגה example-project.us-central1.oracle-view

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