Desarrollar un conector personalizado para importar metadatos

En este documento se proporciona una plantilla de referencia para crear un conector personalizado que extraiga metadatos de fuentes de terceros, como MySQL, SQL Server y Oracle. Puede usar este conector para importar metadatos a Dataplex Universal Catalog a través de una pipeline de conectividad gestionada. Se incluye un conector de Python de ejemplo para Oracle Database Express Edition (XE) como punto de partida. También puedes desarrollar conectores con Java, Scala o R.

Cómo funcionan los conectores

Un conector extrae metadatos de una fuente de datos de terceros, los transforma al formato ImportItem de Dataplex Universal Catalog y genera archivos de importación de metadatos que puede importar Dataplex Universal Catalog.

El conector forma parte de una canalización de conectividad gestionada. Una canalización de conectividad gestionada es un flujo de trabajo orquestado que se usa para importar metadatos de Dataplex Universal Catalog. La canalización de conectividad gestionada ejecuta el conector y realiza otras tareas en el flujo de trabajo de importación, como ejecutar un trabajo de importación de metadatos y registrar los registros.

La canalización de conectividad gestionada ejecuta el conector mediante un trabajo por lotes de Google Cloud Serverless para Apache Spark. Serverless for Apache Spark proporciona un entorno de ejecución de Spark sin servidor. Aunque puedes crear un conector que no use Spark, te recomendamos que lo uses porque puede mejorar el rendimiento de tu conector.

Requisitos de los conectores

El conector tiene los siguientes requisitos:

• El conector debe ser una imagen de Artifact Registry que se pueda ejecutar en Serverless para Apache Spark.
• El conector debe generar archivos de metadatos en un formato que se pueda importar mediante una tarea de importación de metadatos de Dataplex Universal Catalog (el método de la API metadataJobs.create). Para obtener información detallada sobre los requisitos, consulta el artículo Archivo de importación de metadatos.

• El conector debe aceptar los siguientes argumentos de línea de comandos para recibir información de la canalización:

  Argumento de línea de comandos	Valor que proporciona la canalización
  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

  El conector usa estos argumentos para generar metadatos en un grupo de entradas de destino projects/PROJECT_ID/locations/REGION/entryGroups/ENTRY_GROUP_ID y para escribir en un segmento de Cloud Storage gs://CLOUD_STORAGE_BUCKET_ID/FOLDER_ID. Cada vez que se ejecuta la canalización, se crea una carpeta FOLDER_ID en el segmento CLOUD_STORAGE_BUCKET_ID. El conector debe escribir archivos de importación de metadatos en esta carpeta.

Las plantillas de canalización admiten conectores de PySpark. Las plantillas asumen que el controlador (mainPythonFileUri) es un archivo local de la imagen del conector llamado main.py. Puede modificar las plantillas de canalización para otros casos, como un conector de Spark, un URI de controlador diferente u otras opciones.

A continuación, se explica cómo usar PySpark para crear un elemento de importación en el archivo de importación de metadatos.

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

Antes de empezar

En esta guía se da por supuesto que tienes conocimientos sobre Python y PySpark.

Revisa la siguiente información:

• Conceptos de metadatos de Dataplex Universal Catalog
• Documentación sobre los trabajos de importación de metadatos

Haz lo siguiente. Crea todos los recursos en la misma Google Cloud ubicación.

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

2. Verify that billing is enabled for your Google Cloud project.

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

4. Install the Google Cloud CLI.

5. Si utilizas un proveedor de identidades (IdP) externo, primero debes iniciar sesión en la CLI de gcloud con tu identidad federada.

6. Para inicializar gcloud CLI, ejecuta el siguiente comando:

   gcloud init

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

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

9. Crea un segmento de Cloud Storage para almacenar los archivos de importación de metadatos.

10. Crea los siguientes recursos de metadatos en el mismo proyecto.

    Para ver valores de ejemplo, consulta la sección Recursos de metadatos de ejemplo para una fuente de Oracle de este documento.

    1. Crea un grupo de entradas.

    2. Crea tipos de aspectos personalizados para las entradas que quieras importar. Usa la convención de nomenclatura SOURCE-ENTITY_TO_IMPORT.

       Por ejemplo, en una base de datos Oracle, crea un tipo de aspecto llamado oracle-database.

       Si quiere, puede crear tipos de aspectos adicionales para almacenar otra información.

    3. Crea tipos de entrada personalizados para los recursos que quieras importar y asígnales los tipos de aspecto correspondientes. Usa la convención de nomenclatura SOURCE-ENTITY_TO_IMPORT.

       Por ejemplo, en una base de datos Oracle, crea un tipo de entrada llamado oracle-database. Vincúlalo al tipo de aspecto llamado oracle-database.

11. Asegúrate de que se pueda acceder a tu fuente de terceros desde tu Google Cloud proyecto. Para obtener más información, consulta Configuración de red de Serverless para Apache Spark.

Crear un conector básico de Python

El conector básico de Python de ejemplo crea entradas de nivel superior para una fuente de datos de Oracle mediante las clases de la biblioteca de cliente de Universal Catalog de Dataplex. A continuación, proporciona los valores de los campos de entrada.

El conector crea un archivo de importación de metadatos con las siguientes entradas:

• Una entrada instance, con el tipo de entrada projects/PROJECT_ID/locations/LOCATION/entryTypes/oracle-instance. Esta entrada representa un sistema Oracle Database XE.
• Una entrada database, que representa una base de datos dentro del sistema Oracle Database XE.

Para crear un conector básico de Python, sigue estos pasos:

1. Clona el cloud-dataplexrepositorio.

2. Configura un entorno local. Te recomendamos que uses un entorno virtual.

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

   Usa las versiones activas o de mantenimiento de Python. Se admiten las versiones de Python 3.7 y posteriores.

3. Crea un proyecto de Python.

4. Requisitos de instalación:

   pip install -r requirements.txt
   

   Se han instalado los siguientes requisitos:

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

5. Añade un archivo de canalización main.py en la raíz del proyecto.

   from src import bootstrap
   
   
   if __name__ == '__main__':
       bootstrap.run()
   

   Al implementar tu código en Serverless para Apache Spark, el archivo main.py sirve como punto de entrada para la ejecución. Te recomendamos que minimices la cantidad de información que se almacena en el archivo main.py. Utiliza este archivo para llamar a funciones y clases que se definan en tu conector, como la clase src/bootstap.py.

6. Crea una carpeta src para almacenar la mayor parte de la lógica de tu conector.

7. Actualiza el archivo src/cmd_reader.py con una clase de Python para aceptar argumentos de línea de comandos. Para ello, puedes usar el módulo 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])
   

   En entornos de producción, te recomendamos que almacenes la contraseña en Secret Manager.

8. Actualiza el archivo src/constants.py con el código para crear constantes.

   """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. Actualiza el archivo src/name_builder.py con métodos para crear los recursos de metadatos que quieras que cree el conector para tus recursos de Oracle. Sigue las convenciones que se describen en la sección Recursos de metadatos de ejemplo para una fuente de Oracle de este documento.

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

   Como el archivo name_builder.py se usa tanto para el código principal de Python como para el código principal de PySpark, te recomendamos que escribas los métodos como funciones puras en lugar de como miembros de una clase.

10. Actualiza el archivo src/top_entry_builder.py con el código para rellenar las entradas de nivel superior con datos.

    """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. Actualiza el archivo src/bootstrap.py con el código para generar el archivo de importación de metadatos y ejecuta el conector.

    """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. Ejecuta el código de forma local.

    Se devuelve un archivo de importación de metadatos llamado output.jsonl. El archivo tiene dos líneas, cada una de las cuales representa un elemento de importación. La canalización de conectividad gestionada lee este archivo al ejecutar la tarea de importación de metadatos.

13. Opcional: Amplía el ejemplo anterior para usar las clases de la biblioteca de cliente de Dataplex Universal Catalog y crear elementos de importación para tablas, esquemas y vistas. También puedes ejecutar el ejemplo de Python en Serverless para Apache Spark.

    Te recomendamos que crees un conector que use Spark (y que se ejecute en Serverless para Apache Spark), ya que puede mejorar el rendimiento del conector.

Crear un conector PySpark

Este ejemplo se basa en la API DataFrame de PySpark. Puedes instalar PySpark SQL y ejecutarlo localmente antes de hacerlo en Serverless para Apache Spark. Si instalas y ejecutas PySpark de forma local, instala la biblioteca PySpark con pip, pero no es necesario que instales un clúster de Spark local.

Por motivos de rendimiento, en este ejemplo no se usan clases predefinidas de la biblioteca PySpark. En su lugar, el ejemplo crea DataFrames, los convierte en entradas JSON y, a continuación, escribe la salida en un archivo de importación de metadatos en formato JSON Lines que se puede importar en el catálogo universal de Dataplex.

Para crear un conector con PySpark, sigue estos pasos:

1. Clona el cloud-dataplexrepositorio.

2. Instala PySpark:

   pip install pyspark
   

3. Requisitos de instalación:

   pip install -r requirements.txt
   

   Se han instalado los siguientes requisitos:

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

4. Actualiza el archivo oracle_connector.py con código para leer datos de una fuente de datos de Oracle y devolver 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