메타데이터 가져오기를 위한 커스텀 커넥터 개발

이 문서에서는 서드 파티 소스에서 메타데이터를 추출하는 커스텀 커넥터를 빌드할 수 있는 참조 템플릿을 제공합니다. 메타데이터를 Dataplex Universal Catalog로 가져오는 관리형 연결 파이프라인을 실행할 때 이 커넥터를 사용합니다.

커넥터를 빌드하여 서드 파티 소스에서 메타데이터를 추출할 수 있습니다. 예를 들어 MySQL, SQL Server, Oracle, Snowflake, Databricks와 같은 소스에서 데이터를 추출하는 커넥터를 빌드할 수 있습니다.

이 문서의 커넥터 예시를 시작점으로 사용하여 자체 커넥터를 빌드하세요. 이 커넥터 예시는 Oracle Database Express Edition(XE) 데이터베이스에 연결합니다. 커넥터는 Python으로 빌드되지만 Java, Scala, R을 사용할 수도 있습니다.

커넥터의 작동 원리

커넥터는 서드 파티 데이터 소스에서 메타데이터를 추출하고 메타데이터를 Dataplex Universal Catalog ImportItem 형식으로 변환하며 Dataplex Universal Catalog에서 가져올 수 있는 메타데이터 가져오기 파일을 만듭니다.

커넥터는 관리형 연결 파이프라인의 일부입니다. 관리형 연결 파이프라인은 개발자가 Dataplex Universal Catalog 메타데이터를 가져오는 데 사용하는 조정된 워크플로입니다. 관리형 연결 파이프라인은 커넥터를 실행하고 메타데이터 가져오기 작업 실행, 로그 캡처와 같은 가져오기 워크플로의 다른 작업을 실행합니다.

관리형 연결 파이프라인은 Google Cloud Apache Spark용 서버리스 일괄 작업을 사용하여 커넥터를 실행합니다. Apache Spark용 서버리스는 서버리스 Spark 실행 환경을 제공합니다. Spark를 사용하지 않는 커넥터를 빌드할 수 있지만 커넥터의 성능을 개선할 수 있으므로 Spark를 사용하는 것이 좋습니다.

커넥터 요구사항

커넥터의 요구사항은 다음과 같습니다.

커넥터는 Apache Spark용 서버리스에서 실행할 수 있는 Artifact Registry 이미지여야 합니다.
커넥터는 Dataplex Universal Catalog 메타데이터 가져오기 작업(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에 씁니다. 파이프라인을 실행할 때마다 버킷 CLOUD_STORAGE_BUCKET_ID에 새 폴더 FOLDER_ID가 생성됩니다. 커넥터는 이 폴더에 메타데이터 가져오기 파일을 작성해야 합니다.

파이프라인 템플릿은 PySpark 커넥터를 지원합니다. 템플릿은 드라이버(mainPythonFileUri)가 커넥터 이미지 main.py의 로컬 파일이라고 가정합니다. 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 및 PySpark에 익숙하다고 가정합니다.

다음 정보를 검토하세요.

Dataplex Universal Catalog 메타데이터 개념
메타데이터 가져오기 작업에 관한 문서

다음 단계를 따르세요. 모든 리소스를 동일한 Google Cloud위치에 만듭니다.

Create or select a Google Cloud project.
Roles required to select or create a project
- Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
- Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
- Create a Google Cloud project:
```
gcloud projects create PROJECT_ID
```
  Replace PROJECT_ID with a name for the Google Cloud project you are creating.
- Select the Google Cloud project that you created:
```
gcloud config set project PROJECT_ID
```
  Replace PROJECT_ID with your Google Cloud project name.
Verify that billing is enabled for your Google Cloud project.
Enable the Dataplex, Dataproc, Workflows, and Artifact Registry APIs:

Roles required to enable APIs
To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.
```
gcloud services enable dataplex.googleapis.com dataproc.googleapis.com workflows.googleapis.com artifactregistry.googleapis.com
```

Install the Google Cloud CLI.
외부 ID 프로바이더(IdP)를 사용하는 경우 먼저 제휴 ID로 gcloud CLI에 로그인해야 합니다.
gcloud CLI를 초기화하려면, 다음 명령어를 실행합니다.
```
gcloud init
```
Grant roles to your user account. Run the following command once for each of the following IAM roles: 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
```
Replace the following:
- PROJECT_ID: Your project ID.
- USER_IDENTIFIER: The identifier for your user account. For example, myemail@example.com.
- ROLE: The IAM role that you grant to your user account.

Set up authentication:
1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator) and the Project IAM Admin role (roles/resourcemanager.projectIamAdmin). Learn how to grant roles.
2. Create the service account:
```
gcloud iam service-accounts create SERVICE_ACCOUNT_NAME
```
  Replace SERVICE_ACCOUNT_NAME with a name for the service account.
3. Grant the roles/owner IAM role to the service account:
```
gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com" --role=roles/owner
```
  Replace the following:
  - SERVICE_ACCOUNT_NAME: the name of the service account
  - PROJECT_ID: the project ID where you created the service account
  Note: The --role flag affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later. In production environments, do not grant the Owner, Editor, or Viewer roles. Instead, grant a predefined role or custom role that meets your needs.

메타데이터 가져오기 파일을 저장할 Cloud Storage 버킷을 만듭니다.
같은 프로젝트에서 다음 메타데이터 리소스를 만듭니다.

참고: 이 문서에서는 편의를 위해 모든 리소스가 동일한 프로젝트에 있다고 가정합니다. 다른 프로젝트에서 리소스를 만들려면 샘플 코드를 조정하면 됩니다.

값의 예시는 이 문서의 Oracle 소스의 메타데이터 리소스 예시 섹션을 참조하세요.
1. 항목 그룹을 만듭니다.
2. 가져오려는 항목의 커스텀 관점 유형을 만듭니다. SOURCE-ENTITY_TO_IMPORT 이름 지정 규칙을 사용합니다.
  
  예를 들어 Oracle 데이터베이스의 경우 oracle-database 관점 유형을 만듭니다.
  
  원하는 경우 다른 정보를 저장하기 위해 관점 유형을 추가로 만들 수 있습니다.
3. 가져오려는 리소스의 커스텀 항목 유형을 만들고 관련 관점 유형을 할당합니다. SOURCE-ENTITY_TO_IMPORT 이름 지정 규칙을 사용합니다.
  
  예를 들어 Oracle 데이터베이스의 경우 oracle-database 항목 유형을 만듭니다. oracle-database이라는 관점 유형에 연결합니다.
Google Cloud 프로젝트에서 서드 파티 소스에 액세스할 수 있는지 확인합니다. 자세한 내용은 Apache Spark용 서버리스 네트워크 구성을 참조하세요.

기본 Python 커넥터 만들기

이 기본 Python 커넥터 예시에서는 Dataplex Universal Catalog 클라이언트 라이브러리 클래스를 사용하여 Oracle 데이터 소스의 최상위 항목을 만듭니다. 그런 다음 입력란에 값을 제공합니다.

커넥터는 다음 항목이 포함된 메타데이터 가져오기 파일을 만듭니다.

항목 유형이 projects/PROJECT_ID/locations/LOCATION/entryTypes/oracle-instance인 instance 항목. 이 항목은 Oracle Database XE 시스템을 나타냅니다.
Oracle Database XE 시스템에 있는 데이터베이스를 나타내는 database 항목

기본 Python 커넥터를 빌드하려면 다음을 수행합니다.

cloud-dataplex 저장소를 클론합니다.
로컬 환경을 설정합니다. 가상 환경을 사용하는 것이 좋습니다.
```
mkdir venv
python -m venv venv/
source venv/bin/activate
```
Python의 활성 또는 유지보수 버전을 사용합니다. Python 버전 3.7 이상이 지원됩니다.
Python 프로젝트를 만듭니다.

설치 요구사항:

pip install -r requirements.txt

다음 요구사항이 설치됩니다.

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

프로젝트 루트에 main.py 파이프라인 파일을 추가합니다.
```
from src import bootstrap


if __name__ == '__main__':
    bootstrap.run()
```
Apache Spark용 서버리스에 코드를 배포할 때 main.py 파일이 실행 진입점 역할을 합니다. main.py 파일에 저장되는 정보의 양을 최소화하는 것이 좋습니다. 이 파일을 사용하여 커넥터 내에 정의된 함수와 클래스(예: src/bootstap.py 클래스)를 호출합니다.
커넥터의 로직 대부분을 저장할 src 폴더를 만듭니다.

명령줄 인수를 허용하도록 Python 클래스로 src/cmd_reader.py 파일을 업데이트합니다. 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에 비밀번호를 저장하는 것이 좋습니다.

상수를 만드는 코드로 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"

커넥터가 Oracle 리소스에 만들려는 메타데이터 리소스를 빌드하는 메서드로 src/name_builder.py 파일을 업데이트합니다. 이 문서의 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 핵심 코드 모두에 사용되므로 메서드를 클래스의 구성원이 아닌 순수 함수로 작성하는 것이 좋습니다.

최상위 항목을 데이터로 채우는 코드로 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))

메타데이터 가져오기 파일을 생성하고 커넥터를 실행하는 코드로 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