Importar metadatos de una fuente personalizada con Workflows

En este documento se describe cómo importar metadatos de fuentes de terceros a Dataplex Universal Catalog configurando y ejecutando una canalización de conectividad gestionada en Workflows. Esta canalización extrae metadatos de tu fuente de datos personalizada y los importa a Universal Catalog de Dataplex, lo que crea los grupos de entradas necesarios.

Para obtener más información sobre la conectividad gestionada, consulta el artículo Información general sobre la conectividad gestionada.

Antes de empezar

Antes de importar metadatos, completa las tareas de esta sección.

Crear un conector

Un conector extrae los metadatos de su fuente de datos y genera un archivo de importación de metadatos que puede importar Dataplex Universal Catalog. El conector es una imagen de Artifact Registry que se puede ejecutar en Google Cloud Serverless para Apache Spark.

Configurar recursos de Google Cloud

  1. Enable the Workflows, Dataproc, Cloud Storage, Dataplex, Secret Manager, Artifact Registry, and Cloud Scheduler 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.

    Enable the APIs

    Si no tienes previsto ejecutar la canalización de forma programada, no es necesario que habilites la API Cloud Scheduler.

  2. Crea secretos en Secret Manager para almacenar las credenciales de tu fuente de datos de terceros.

  3. Configura tu red de nube privada virtual (VPC) para ejecutar cargas de trabajo de Serverless para Apache Spark.

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

  5. Crea los siguientes recursos de Dataplex Universal Catalog:

    1. Crea tipos de aspecto personalizados para las entradas que quieras importar.

    2. Crea tipos de entrada personalizados para las entradas que quieras importar.

Roles obligatorios

Una cuenta de servicio representa la identidad de un flujo de trabajo y determina qué permisos tiene el flujo de trabajo y a qué recursos puede acceder. Google Cloud Necesitas una cuenta de servicio para Workflows (para ejecutar la canalización) y otra para Serverless para Apache Spark (para ejecutar el conector).

Puedes usar la cuenta de servicio predeterminada de Compute Engine (PROJECT_NUMBER-compute@developer.gserviceaccount.com) o crear tu propia cuenta de servicio (o cuentas) para ejecutar la canalización de conectividad gestionada.

Consola

  1. En la consola de Google Cloud , ve a la página Gestión de identidades y accesos.

    Ir a IAM

  2. Selecciona el proyecto en el que quieras importar los metadatos.

  3. Haz clic en Dar acceso. A continuación, introduce la dirección de correo de la cuenta de servicio.

  4. Asigna los siguientes roles a la cuenta de servicio:

    • Editor de registros
    • Propietario de EntryGroup de Dataplex
    • Propietario de trabajos de metadatos de Dataplex
    • Editor de catálogo de Dataplex
    • Editor de Dataproc
    • Trabajador de Dataproc
    • Lector de recursos de Secret Manager en el secreto que almacena las credenciales de tu fuente de datos
    • Usuario de objetos de almacenamiento en el segmento de Cloud Storage
    • Lector de Artifact Registry: en el repositorio de Artifact Registry que contiene la imagen del conector
    • Usuario de cuenta de servicio: si usas diferentes cuentas de servicio, asigna este rol a la cuenta de servicio que ejecuta Workflows en la cuenta de servicio que ejecuta los trabajos por lotes de Serverless para Apache Spark.
    • Invocador de flujos de trabajo: si quieres programar la canalización

  5. Guarda los cambios.

gcloud

  1. Asigna roles a la cuenta de servicio. Ejecuta estos comandos:

    gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/logging.logWriter
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataplex.entryGroupOwner
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataplex.metadataJobOwner
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataplex.catalogEditor
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataproc.editor
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataproc.worker

    Haz los cambios siguientes:

    • PROJECT_ID: el nombre del proyecto Google Cloud de destino al que se importarán los metadatos.
    • SERVICE_ACCOUNT_ID: la cuenta de servicio, como my-service-account@my-project.iam.gserviceaccount.com.

  2. Concede a la cuenta de servicio los siguientes roles a nivel de recurso:

    gcloud secrets add-iam-policy-binding SECRET_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/secretmanager.secretaccessor
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/storage.objectUser \
    --condition=resource.name.startsWith('projects/_/buckets/BUCKET_ID')
gcloud artifacts repositories add-iam-policy-binding REPOSITORY \
    --location=REPOSITORY_LOCATION \
    --member=SERVICE_ACCOUNT_ID} \
    --role=roles/artifactregistry.reader

    Haz los cambios siguientes:

    • SECRET_ID: el ID del secreto que almacena las credenciales de su fuente de datos. Utiliza el formato projects/PROJECT_ID/secrets/SECRET_ID.
    • BUCKET_ID: el nombre del segmento de Cloud Storage.
    • REPOSITORY: el repositorio de Artifact Registry que contiene la imagen del conector.
    • REPOSITORY_LOCATION: la Google Cloud ubicación en la que se aloja el repositorio.

  3. Asigna a la cuenta de servicio que ejecuta Workflows el rol roles/iam.serviceAccountUser en la cuenta de servicio que ejecuta los trabajos por lotes de Serverless para Apache Spark. Debes conceder este rol aunque uses la misma cuenta de servicio para Workflows y Serverless para Apache Spark.

    gcloud iam service-accounts add-iam-policy-binding \
    serviceAccount:SERVICE_ACCOUNT_ID \
    --member='SERVICE_ACCOUNT_ID' \
    --role='roles/iam.serviceAccountUser'

    Si usas cuentas de servicio diferentes, el valor de la marca --member es la cuenta de servicio que ejecuta los trabajos por lotes de Serverless para Apache Spark.

  4. Si quieres programar la canalización, concede a la cuenta de servicio el siguiente rol:

    gcloud projects add-iam-policy-binding PROJECT_ID \
 --member="SERVICE_ACCOUNT_ID" \
 --role=roles/workflows.invoker

Importar metadatos

Para importar metadatos, crea y ejecuta un flujo de trabajo que ejecute la canalización de conectividad gestionada. También puedes crear una programación para ejecutar la canalización.

Consola

  1. Crea el flujo de trabajo. Introduce la información siguiente:

    • Cuenta de servicio: la cuenta de servicio que has configurado en la sección Roles obligatorios de este documento.

    • Cifrado: selecciona Google-managed encryption key.

    • Define el flujo de trabajo: proporciona el siguiente archivo de definición:

      main:
  params: [args]
  steps:
    - init:
        assign:
        - WORKFLOW_ID: ${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
        - NETWORK_URI: ${default(map.get(args, "NETWORK_URI"), "")}
        - SUBNETWORK_URI: ${default(map.get(args, "SUBNETWORK_URI"), "")}
        - NETWORK_TAGS: ${default(map.get(args, "NETWORK_TAGS"), [])}

    - check_networking:
        switch:
          - condition: ${NETWORK_URI != "" and SUBNETWORK_URI != ""}
            raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
          - condition: ${NETWORK_URI == "" and SUBNETWORK_URI == ""}
            steps:
             - submit_extract_job_with_default_network_uri:
                  assign:
                    - NETWORK_TYPE: "networkUri"
                    - NETWORKING: ${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}  
          - condition: ${NETWORK_URI != ""}
            steps:
              - submit_extract_job_with_network_uri:
                  assign:
                    - NETWORKING: ${NETWORK_URI}
                    - NETWORK_TYPE: "networkUri"
          - condition: ${SUBNETWORK_URI != ""}
            steps:
              - submit_extract_job_with_subnetwork_uri:
                  assign:
                    - NETWORKING: ${SUBNETWORK_URI}
                    - NETWORK_TYPE: "subnetworkUri"
        next: check_create_target_entry_group

    - check_create_target_entry_group:
        switch:
          - condition: ${args.CREATE_TARGET_ENTRY_GROUP == true}
            next: create_target_entry_group
          - condition: ${args.CREATE_TARGET_ENTRY_GROUP == false}
            next: prepare_pyspark_job_body

    - create_target_entry_group:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: prepare_pyspark_job_body

    - prepare_pyspark_job_body:
        assign:
          - pyspark_batch_body:
              mainPythonFileUri: file:///main.py
              args: