Desarrolla un conector personalizado para la importación de metadatos

En este documento, se proporciona una plantilla de referencia para compilar un conector personalizado que permita extraer metadatos de fuentes externas, como MySQL, SQL Server y Oracle. Puedes usar este conector para importar metadatos a Knowledge Catalog (anteriormente, Dataplex Universal Catalog) a través de una canalización de conectividad administrada. Se incluye un ejemplo de conector de Python 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 externa, los transforma al formato ImportItem de Knowledge Catalog y genera archivos de importación de metadatos que Knowledge Catalog puede importar.

El conector forma parte de una canalización de conectividad administrada. Una canalización de conectividad administrada es un flujo de trabajo organizado que se usa para importar metadatos del Knowledge Catalog. La canalización de conectividad administrada 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 capturar registros.

La canalización de conectividad administrada ejecuta el conector con un trabajo por lotes de Managed Service para Apache Spark. Managed Service para Apache Spark proporciona un entorno de ejecución de Spark sin servidores. Si bien puedes compilar un conector que no use Spark, te recomendamos que lo uses porque puede mejorar el rendimiento de tu conector.

Requisitos del conector

El conector tiene los siguientes requisitos:

  • El conector debe ser una imagen de Artifact Registry que se pueda ejecutar en Managed Service para Apache Spark.
  • El conector debe generar archivos de metadatos en un formato que pueda importar un trabajo de importación de metadatos de Knowledge Catalog (el método de la API de metadataJobs.create). Para conocer los requisitos detallados, consulta 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 la 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 entrada de destino projects/PROJECT_ID/locations/REGION/entryGroups/ENTRY_GROUP_ID y para escribir en un bucket de Cloud Storage gs://CLOUD_STORAGE_BUCKET_ID/FOLDER_ID. Cada ejecución de la canalización crea una carpeta nueva FOLDER_ID en el bucket 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 suponen que el controlador (mainPythonFileUri) es un archivo local en la imagen del conector llamado main.py. Puedes modificar las plantillas de canalización para otros casos, como un conector de Spark, un URI de controlador diferente o cualquier otra opción.

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 comenzar

En esta guía, se supone que conoces Python y PySpark.

Revisa la siguiente información:

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

  1. Crea o selecciona un Google Cloud proyecto.

    Roles necesarios para seleccionar o crear un proyecto

    • Selecciona un proyecto: Para seleccionar un proyecto, no se requiere un rol de IAM específico. Puedes seleccionar cualquier proyecto en el que se te haya otorgado un rol.
    • Crear un proyecto: Para crear un proyecto, necesitas el rol de Creador de proyectos (roles/resourcemanager.projectCreator), que contiene el permiso resourcemanager.projects.create. Obtén más información para otorgar roles.

    • Crea un Google Cloud proyecto:

      gcloud projects create PROJECT_ID

      Reemplaza