Google usa tecnología de IA para traducir contenido al idioma que prefieras. Las traducciones generadas por IA pueden contener errores.

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.

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.
Si utilizas un proveedor de identidades (IdP) externo, primero debes iniciar sesión en la CLI de gcloud con tu identidad federada.
Para inicializar gcloud CLI, ejecuta el siguiente comando:
```
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.

Crea un segmento de Cloud Storage para almacenar los archivos de importación de metadatos.
Crea los siguientes recursos de metadatos en el mismo proyecto.

Nota: En este documento se da por supuesto que todos los recursos están en el mismo proyecto para simplificar las cosas. Si quieres crear recursos en proyectos diferentes, puedes ajustar el código de ejemplo.

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