Mengembangkan konektor kustom untuk impor metadata

Dokumen ini menyediakan template referensi untuk membuat konektor kustom guna mengekstrak metadata dari sumber pihak ketiga, seperti MySQL, SQL Server, dan Oracle. Anda dapat menggunakan konektor ini untuk mengimpor metadata ke Dataplex Universal Catalog melalui pipeline konektivitas terkelola. Contoh konektor Python untuk Oracle Database Express Edition (XE) disertakan sebagai titik awal. Anda juga dapat mengembangkan konektor menggunakan Java, Scala, atau R.

Cara kerja konektor

Konektor mengekstrak metadata dari sumber data pihak ketiga, mengubah metadata ke format ImportItem Dataplex Universal Catalog, dan membuat file impor metadata yang dapat diimpor oleh Dataplex Universal Catalog.

Konektor adalah bagian dari pipeline konektivitas terkelola. Pipeline konektivitas terkelola adalah alur kerja yang diatur yang Anda gunakan untuk mengimpor metadata Dataplex Universal Catalog. Pipeline konektivitas terkelola menjalankan konektor dan melakukan tugas lain dalam alur kerja impor, seperti menjalankan tugas impor metadata dan merekam log.

Pipeline konektivitas terkelola menjalankan konektor menggunakan tugas batch Google Cloud Serverless for Apache Spark. Serverless for Apache Spark menyediakan lingkungan eksekusi Spark serverless. Meskipun Anda dapat membuat konektor yang tidak menggunakan Spark, sebaiknya gunakan Spark karena dapat meningkatkan performa konektor Anda.

Persyaratan konektor

Konektor memiliki persyaratan berikut:

  • Konektor harus berupa image Artifact Registry yang dapat dijalankan di Serverless untuk Apache Spark.
  • Konektor harus membuat file metadata dalam format yang dapat diimpor oleh tugas impor metadata Dataplex Universal Catalog (metode API metadataJobs.create). Untuk mengetahui persyaratan mendetail, lihat File impor metadata.

  • Konektor harus menerima argumen command line berikut untuk menerima informasi dari pipeline:

    Argumen command line Nilai yang diberikan oleh saluran
    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

    Konektor menggunakan argumen ini untuk membuat metadata dalam grup entri target projects/PROJECT_ID/locations/REGION/entryGroups/ENTRY_GROUP_ID, dan untuk menulis ke bucket Cloud Storage gs://CLOUD_STORAGE_BUCKET_ID/FOLDER_ID. Setiap eksekusi pipeline akan membuat folder baru FOLDER_ID di bucket CLOUD_STORAGE_BUCKET_ID. Konektor harus menulis file impor metadata ke folder ini.

Template pipeline mendukung konektor PySpark. Template mengasumsikan bahwa driver (mainPythonFileUri) adalah file lokal pada image konektor bernama main.py. Anda dapat mengubah template pipeline untuk skenario lain, seperti konektor Spark, URI driver yang berbeda, atau opsi lainnya.

Berikut cara menggunakan PySpark untuk membuat item impor dalam file impor metadata.

"""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()))
])

Sebelum memulai

Panduan ini mengasumsikan bahwa Anda sudah terbiasa dengan Python dan PySpark.

Tinjau informasi berikut:

Lakukan hal-hal berikut. Buat semua resource di lokasi yang sama. Google Cloud

  1. Buat atau pilih Google Cloud project.

    Peran yang diperlukan untuk memilih atau membuat project

    • Pilih project: Memilih project tidak memerlukan peran IAM tertentu—Anda dapat memilih project mana pun yang telah diberi peran.
    • Membuat project: Untuk membuat project, Anda memerlukan peran Pembuat Project (roles/resourcemanager.projectCreator), yang berisi izin resourcemanager.projects.create. Pelajari cara memberikan peran.

    • Buat Google Cloud project:

      gcloud projects create PROJECT_ID

      Ganti PROJECT_ID dengan nama untuk Google Cloud project yang Anda buat.

    • Pilih project Google Cloud yang Anda buat:

      gcloud config set project PROJECT_ID

      Ganti PROJECT_ID dengan nama project Google Cloud Anda.

  2. Verifikasi bahwa penagihan diaktifkan untuk project Google Cloud Anda.

  3. Aktifkan Dataplex, Dataproc, Workflows, dan Artifact Registry API:

    Peran yang diperlukan untuk mengaktifkan API

    Untuk mengaktifkan API, Anda memerlukan peran IAM Service Usage Admin (roles/serviceusage.serviceUsageAdmin), yang berisi izin serviceusage.services.enable. Pelajari cara memberikan peran. 

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

  4. Instal Google Cloud CLI.

  5. Jika Anda menggunakan penyedia identitas (IdP) eksternal, Anda harus login ke gcloud CLI dengan identitas gabungan Anda terlebih dahulu.

  6. Untuk melakukan inisialisasi gcloud CLI, jalankan perintah berikut: 

    gcloud init

  7. Memberikan peran ke akun pengguna Anda. Jalankan perintah berikut satu kali untuk setiap peran IAM berikut: 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

    Ganti kode berikut:

    • PROJECT_ID: Project ID Anda.
    • USER_IDENTIFIER: ID untuk akun pengguna Anda. Misalnya, myemail@example.com.
    • ROLE: Peran IAM yang Anda berikan ke akun pengguna Anda.

  8. Menyiapkan autentikasi:

    1. Pastikan Anda memiliki peran IAM Create Service Accounts (roles/iam.serviceAccountCreator) dan peran IAM Project IAM Admin (roles/resourcemanager.projectIamAdmin). Pelajari cara memberikan peran.

    2. Buat akun layanan: 

      gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

      Ganti SERVICE_ACCOUNT_NAME dengan nama untuk akun layanan.

    3. Memberikan peran IAM roles/owner ke akun layanan. 

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

      Ganti kode berikut:

      • SERVICE_ACCOUNT_NAME: nama dari akun layanan.
      • PROJECT_ID: project ID dimana Anda membuat akun layanan

  9. Buat bucket Cloud Storage untuk menyimpan file impor metadata.

  10. Buat resource metadata berikut dalam project yang sama.

    Untuk nilai contoh, lihat bagian Contoh resource metadata untuk sumber Oracle dalam dokumen ini.

    1. Buat grup entri.

    2. Buat jenis aspek kustom untuk entri yang ingin Anda impor. Gunakan konvensi penamaan SOURCE-ENTITY_TO_IMPORT.

      Misalnya, untuk database Oracle, buat jenis aspek bernama oracle-database.

      Secara opsional, Anda dapat membuat jenis aspek tambahan untuk menyimpan informasi lainnya.

    3. Buat jenis entri kustom untuk resource yang ingin Anda impor, dan tetapkan jenis aspek yang relevan kepadanya. Gunakan konvensi penamaan SOURCE-ENTITY_TO_IMPORT.

      Misalnya, untuk database Oracle, buat jenis entri bernama oracle-database. Tautkan ke jenis aspek yang diberi nama oracle-database.

  11. Pastikan sumber pihak ketiga Anda dapat diakses dari project Google Cloud Anda. Untuk mengetahui informasi selengkapnya, lihat Konfigurasi jaringan Serverless untuk Apache Spark.

Membuat konektor Python dasar

Contoh konektor Python dasar membuat entri tingkat teratas untuk sumber data Oracle menggunakan class library klien Dataplex Universal Catalog. Kemudian, Anda memberikan nilai untuk kolom entri.

Konektor membuat file impor metadata dengan entri berikut:

  • Entri instance, dengan jenis entri projects/PROJECT_ID/locations/LOCATION/entryTypes/oracle-instance. Entri ini mewakili sistem Oracle Database XE.
  • Entri database, yang merepresentasikan database di dalam sistem Oracle Database XE.

Untuk membuat konektor Python dasar, lakukan hal berikut:

  1. Buat clone repositori cloud-dataplex.

  2. Siapkan lingkungan lokal. Sebaiknya gunakan lingkungan virtual.

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

    Gunakan versi Python aktif atau pemeliharaan. Python versi 3.7 dan yang lebih baru didukung.

  3. Buat project Python.

  4. Persyaratan penginstalan:

    pip install -r requirements.txt

    Persyaratan berikut diinstal:

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

  5. Tambahkan file pipeline main.py di root project.

    from src import bootstrap


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

    Saat men-deploy kode ke Serverless for Apache Spark, file main.py berfungsi sebagai titik entri untuk eksekusi. Sebaiknya minimalkan jumlah informasi yang disimpan dalam file main.py; gunakan file ini untuk memanggil fungsi dan class yang ditentukan dalam konektor Anda, seperti class src/bootstap.py.

  6. Buat folder src untuk menyimpan sebagian besar logika konektor Anda.

  7. Perbarui file src/cmd_reader.py dengan class Python untuk menerima argumen command line. Anda dapat menggunakan modul argeparse untuk melakukannya.

    """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])

    Di lingkungan produksi, sebaiknya Anda menyimpan sandi di Secret Manager.

  8. Perbarui file src/constants.py dengan kode untuk membuat konstanta.

    """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. Perbarui file src/name_builder.py dengan metode untuk membangun resource metadata yang Anda inginkan agar konektor dibuat untuk resource Oracle Anda. Gunakan konvensi yang dijelaskan di bagian Contoh resource metadata untuk sumber Oracle dalam dokumen ini.

    """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}"

    Karena file name_builder.py digunakan untuk kode inti Python dan kode inti PySpark, sebaiknya tulis metode sebagai fungsi murni, bukan sebagai anggota class.

  10. Perbarui file src/top_entry_builder.py dengan kode untuk mengisi entri tingkat teratas dengan data.

    """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. Perbarui file src/bootstrap.py dengan kode untuk membuat file impor metadata dan menjalankan konektor.

    """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. Jalankan kode secara lokal.

    File impor metadata bernama output.jsonl ditampilkan. File memiliki dua baris, yang masing-masing mewakili item impor. Pipeline konektivitas terkelola membaca file ini saat menjalankan tugas impor metadata.

  13. Opsional: Perluas contoh sebelumnya untuk menggunakan class library klien Dataplex Universal Catalog guna membuat item impor untuk tabel, skema, dan tampilan. Anda juga dapat menjalankan contoh Python di Serverless for Apache Spark.

    Sebaiknya buat konektor yang menggunakan Spark (dan berjalan di Serverless untuk Apache Spark), karena dapat meningkatkan performa konektor Anda.

Membuat konektor PySpark

Contoh ini didasarkan pada PySpark DataFrame API. Anda dapat menginstal PySpark SQL dan menjalankannya secara lokal sebelum menjalankan di Serverless for Apache Spark. Jika Anda menginstal dan menjalankan PySpark secara lokal, instal library PySpark menggunakan pip, tetapi Anda tidak perlu menginstal cluster Spark lokal.

Untuk alasan performa, contoh ini tidak menggunakan class yang telah ditentukan sebelumnya dari library PySpark. Sebagai gantinya, contoh ini membuat DataFrame, mengonversi DataFrame menjadi entri JSON, lalu menulis output ke file impor metadata dalam format JSON Lines yang dapat diimpor ke Dataplex Universal Catalog.

Untuk membuat konektor menggunakan PySpark, lakukan langkah berikut:

  1. Buat clone repositori cloud-dataplex.

  2. Instal PySpark:

    pip install pyspark

  3. Persyaratan penginstalan:

    pip install -r requirements.txt

    Persyaratan berikut diinstal:

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

  4. Perbarui file oracle_connector.py dengan kode untuk membaca data dari sumber data Oracle dan menampilkan DataFrame.

    """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)

    Tambahkan kueri SQL untuk menampilkan metadata yang ingin Anda impor. Kueri harus menampilkan informasi berikut:

    • Skema database
    • Tabel yang termasuk dalam skema ini
    • Kolom yang termasuk dalam tabel ini, termasuk nama kolom, jenis data kolom, dan apakah kolom dapat bernilai null atau wajib diisi

    Semua kolom dari semua tabel dan tampilan disimpan dalam tabel sistem yang sama. Anda dapat memilih kolom dengan metode _get_columns. Bergantung pada parameter yang Anda berikan, Anda dapat memilih kolom untuk tabel atau untuk tampilan secara terpisah.

    Perhatikan hal berikut:

    • Di Oracle, skema database dimiliki oleh pengguna database dan memiliki nama yang sama dengan pengguna tersebut.
    • Objek skema adalah struktur logis yang dibuat oleh pengguna. Objek seperti tabel atau indeks dapat menyimpan data, dan objek seperti tampilan atau sinonim hanya terdiri dari definisi.
    • File ojdbc11.jar berisi driver JDBC Oracle.

  5. Perbarui file src/entry_builder.py dengan metode bersama untuk menerapkan transformasi 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

    Perhatikan hal berikut:

    • Metode ini membangun resource metadata yang dibuat konektor untuk resource Oracle Anda. Gunakan konvensi yang dijelaskan di bagian Contoh resource metadata untuk sumber Oracle dalam dokumen ini.
    • Metode convert_to_import_items berlaku untuk skema, tabel, dan tampilan. Pastikan output konektor adalah satu atau beberapa item impor yang dapat diproses oleh metode metadataJobs.create, bukan entri individual.
    • Bahkan dalam tampilan, kolom ini disebut TABLE_NAME.

  6. Perbarui file bootstrap.py dengan kode untuk membuat file impor metadata dan menjalankan konektor.

    """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)

    Contoh ini menyimpan file impor metadata sebagai satu file JSON Lines. Anda dapat menggunakan alat PySpark seperti class DataFrameWriter untuk menghasilkan batch JSON secara paralel.

    Konektor dapat menulis entri ke file impor metadata dalam urutan apa pun.

  7. Perbarui file gcs_uploader.py dengan kode untuk mengupload file impor metadata ke bucket 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. Buat image konektor.

    Jika konektor Anda berisi beberapa file, atau jika Anda ingin menggunakan library yang tidak disertakan dalam image Docker default, Anda harus menggunakan container kustom. Serverless for Apache Spark menjalankan workload dalam container Docker. Buat image Docker kustom konektor dan simpan image di Artifact Registry. Serverless for Apache Spark membaca image dari Artifact Registry.

    1. Buat Dockerfile:

      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

      Gunakan Conda sebagai pengelola paket Anda. Serverless untuk Apache Spark memasang pyspark ke dalam container saat runtime, sehingga Anda tidak perlu menginstal dependensi PySpark di image container kustom Anda.

    2. Bangun image container kustom dan kirimkan ke 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}"

      Karena satu image dapat memiliki beberapa nama, Anda dapat menggunakan tag Docker untuk menetapkan alias ke image.

  9. Jalankan konektor di Serverless untuk Apache Spark. Untuk mengirimkan tugas batch PySpark menggunakan image container kustom, jalankan perintah 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

    Perhatikan hal berikut:

    • File JAR adalah driver untuk Spark. Untuk membaca dari Oracle, MySQL, atau Postgres, Anda harus memberikan paket tertentu ke Apache Spark. Paket dapat berada di Cloud Storage atau di dalam container. Jika file JAR berada di dalam penampung, jalur akan mirip dengan file:///path/to/file/driver.jar. Dalam contoh ini, jalur ke file JAR adalah /opt/spark/jars/.
    • PIPELINE_ARGUMENTS adalah argumen command line untuk konektor.

    Konektor mengekstrak metadata dari database Oracle, membuat file impor metadata, dan menyimpan file impor metadata ke bucket Cloud Storage.

  10. Untuk mengimpor metadata secara manual dalam file impor metadata ke Dataplex Universal Catalog, jalankan tugas metadata. Gunakan metode metadataJobs.create.

    1. Di command line, tambahkan variabel lingkungan dan buat alias untuk perintah 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. Panggil metode API, dengan meneruskan jenis entri dan jenis aspek yang ingin Anda impor.

      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
)"

      Jenis aspek schema adalah jenis aspek global yang ditentukan oleh Dataplex Universal Catalog.

      Perhatikan bahwa format yang Anda gunakan untuk nama jenis aspek saat memanggil metode API berbeda dengan format yang Anda gunakan dalam kode konektor.

    3. Opsional: Gunakan Cloud Logging untuk melihat log tugas metadata. Untuk mengetahui informasi selengkapnya, lihat Memantau log Dataplex Universal Catalog.

Menyiapkan orkestrasi pipeline

Bagian sebelumnya menunjukkan cara membuat contoh konektor dan menjalankan konektor secara manual.

Di lingkungan produksi, Anda menjalankan konektor sebagai bagian dari pipeline konektivitas terkelola, dengan menggunakan platform orkestrasi seperti Workflows.

  1. Untuk menjalankan pipeline konektivitas terkelola dengan contoh konektor, ikuti langkah-langkah untuk mengimpor metadata menggunakan Workflows. Lakukan hal berikut:

    • Buat alur kerja di lokasi Google Cloud yang sama dengan konektor.

    • Dalam file definisi alur kerja, perbarui fungsi submit_pyspark_extract_job dengan kode berikut untuk mengekstrak data dari database Oracle menggunakan konektor yang Anda buat.

      - 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.ORACE_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

    • Dalam file definisi alur kerja, perbarui fungsi submit_import_job dengan kode berikut untuk mengimpor entri. Fungsi ini memanggil metode API metadataJobs.create untuk menjalankan tugas impor metadata.

      - 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

      Berikan jenis entri dan jenis aspek yang sama dengan yang Anda sertakan saat memanggil metode API secara manual. Perhatikan bahwa tidak ada koma di akhir setiap string.

    • Saat Anda menjalankan alur kerja, berikan argumen runtime berikut:

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

  2. Opsional: Gunakan Cloud Logging untuk melihat log untuk pipeline konektivitas terkelola. Payload log mencakup link ke log untuk tugas batch Serverless for Apache Spark dan tugas impor metadata, jika relevan. Untuk mengetahui informasi selengkapnya, lihat Melihat log alur kerja.

  3. Opsional: Untuk meningkatkan keamanan, performa, dan fungsi pipeline konektivitas terkelola Anda, pertimbangkan untuk melakukan hal berikut:

    1. Gunakan Secret Manager untuk menyimpan kredensial sumber data pihak ketiga Anda.
    2. Gunakan PySpark untuk menulis output JSON Lines ke beberapa file impor metadata secara paralel.
    3. Gunakan awalan untuk membagi file besar (lebih dari 100 MB) menjadi file yang lebih kecil.
    4. Tambahkan aspek kustom lainnya yang mencakup metadata bisnis dan teknis tambahan dari sumber Anda.

Contoh resource metadata untuk sumber Oracle

Konektor contoh mengekstrak metadata dari database Oracle dan memetakan metadata ke resource metadata Dataplex Universal Catalog yang sesuai.

Pertimbangan hierarki

Setiap sistem di Dataplex Universal Catalog memiliki entri root yang merupakan entri induk untuk sistem tersebut. Biasanya, entri root memiliki jenis entri instance. Tabel berikut menunjukkan contoh hierarki jenis entri dan jenis aspek untuk sistem Oracle. Misalnya, jenis entri oracle-database ditautkan ke jenis aspek yang juga bernama oracle-database.

ID jenis entri Deskripsi ID jenis aspek tertaut
oracle-instance Root sistem yang diimpor. oracle-instance
oracle-database Database Oracle. oracle-database
oracle-schema Skema database. oracle-schema
oracle-table Tabel.

oracle-table

schema
oracle-view Tampilan.

oracle-view

schema

Jenis aspek schema adalah jenis aspek global yang ditentukan oleh Dataplex Universal Catalog. Berisi deskripsi kolom dalam tabel, tampilan, atau entitas lain yang memiliki kolom. Jenis aspek kustom oracle-schema berisi nama skema database Oracle.

Contoh kolom item impor

Konektor harus menggunakan konvensi berikut untuk resource Oracle.

  • Nama yang sepenuhnya memenuhi syarat: nama yang sepenuhnya memenuhi syarat untuk resource Oracle menggunakan template penamaan berikut. Karakter yang dilarang diawali dengan tanda petik terbalik.

    Resource Template Contoh
    Instance

    SOURCE:ADDRESS

    Gunakan host dan nomor port atau nama domain sistem.

    		 oracle:`localhost:1521` atau oracle:`myinstance.com`
    Database SOURCE:ADDRESS.DATABASE oracle:`localhost:1521`.xe
    Skema SOURCE:ADDRESS.DATABASE.SCHEMA oracle:`localhost:1521`.xe.sys
    Tabel SOURCE:ADDRESS.DATABASE.SCHEMA.TABLE_NAME oracle:`localhost:1521`.xe.sys.orders
    Lihat SOURCE:ADDRESS.DATABASE.SCHEMA.VIEW_NAME oracle:`localhost:1521`.xe.sys.orders_view

  • Nama entri atau ID entri: entri untuk resource Oracle menggunakan template penamaan berikut. Karakter yang dilarang diganti dengan karakter yang diizinkan. Resource menggunakan awalan projects/PROJECT/locations/LOCATION/entryGroups/ENTRY_GROUP/entries.

    Resource Template Contoh
    Instance PREFIX/HOST_PORT projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521
    Database PREFIX/HOST_PORT/databases/DATABASE projects/example-project/locations/us-central1/entryGroups/oracle-prod/entries/10.1.1.1@1521/databases/xe
    Skema 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
    Tabel 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
    Lihat 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

  • Entri induk: jika entri bukan entri root untuk sistem, entri dapat memiliki kolom entri induk yang menjelaskan posisinya dalam hierarki. Kolom harus berisi nama entri induk. Sebaiknya Anda membuat nilai ini.

    Tabel berikut menunjukkan entri induk untuk resource Oracle.

    Entri Entri induk
    Instance "" (string kosong)
    Database Nama instance
    Skema Nama database
    Tabel Nama skema
    Lihat Nama skema

  • Peta aspek: peta aspek harus berisi setidaknya satu aspek yang menjelaskan entity yang akan diimpor. Berikut adalah contoh peta aspek untuk tabel Oracle.

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

    Anda dapat menemukan jenis aspek standar (seperti schema) yang menentukan struktur tabel atau tampilan di project dataplex-types, di lokasi global.

  • Kunci aspek: kunci aspek menggunakan format penamaan PROJECT.LOCATION.ASPECT_TYPE. Tabel berikut menunjukkan contoh kunci aspek untuk resource Oracle.

    Entri Contoh kunci aspek
    Instance example-project.us-central1.oracle-instance
    Database example-project.us-central1.oracle-database
    Skema example-project.us-central1.oracle-schema
    Tabel example-project.us-central1.oracle-table
    Lihat example-project.us-central1.oracle-view

Langkah berikutnya