Benutzerdefinierten Connector für den Metadatenimport entwickeln

Dieses Dokument enthält eine Referenzvorlage zum Erstellen eines benutzerdefinierten Connectors zum Extrahieren von Metadaten aus Drittanbieterquellen wie MySQL, SQL Server und Oracle. Mit diesem Connector können Sie Metadaten über eine Pipeline für verwaltete Verbindungen in Knowledge Catalog (ehemals Dataplex Universal Catalog) importieren. Ein Beispiel für einen Python-Connector für Oracle Database Express Edition (XE) ist als Ausgangspunkt enthalten. Sie können Connectors auch mit Java, Scala oder R entwickeln.

Funktionsweise von Connectors

Ein Connector extrahiert Metadaten aus einer Drittanbieterdatenquelle, wandelt die Metadaten in das Knowledge Catalog-Format ImportItem um und generiert Metadaten-Importdateien, die von Knowledge Catalog importiert werden können.

Der Connector ist Teil einer Pipeline für verwaltete Verbindungen. Eine Pipeline für verwaltete Verbindungen ist ein orchestrierter Workflow, mit dem Sie Knowledge Catalog-Metadaten importieren. Die Pipeline für verwaltete Verbindungen führt den Connector aus und übernimmt andere Aufgaben im Importworkflow, z. B. das Ausführen eines Metadatenimportjobs und das Erfassen von Logs.

In der verwalteten Verbindungspipeline wird der Connector mit einem Managed Service for Apache Spark-Batchjob ausgeführt. Managed Service for Apache Spark bietet eine serverlose Spark-Ausführungsumgebung. Sie können zwar einen Connector erstellen, der Spark nicht verwendet, wir empfehlen jedoch, Spark zu verwenden, da dies die Leistung Ihres Connectors verbessern kann.

Anschlussstecker-Anforderungen

Für den Connector gelten die folgenden Anforderungen:

  • Der Connector muss ein Artifact Registry-Image sein, das in Managed Service for Apache Spark ausgeführt werden kann.
  • Der Connector muss Metadatendateien in einem Format generieren, das von einem Knowledge Catalog-Metadatenimportjob (der metadataJobs.create-API-Methode) importiert werden kann. Ausführliche Anforderungen finden Sie unter Metadaten-Importdatei.

  • Der Connector muss die folgenden Befehlszeilenargumente akzeptieren, um Informationen aus der Pipeline zu empfangen:

    Befehlszeilenargument Wert, den die Pipeline bietet
    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

    Der Connector verwendet diese Argumente, um Metadaten in einer Zieleintragsgruppe projects/PROJECT_ID/locations/REGION/entryGroups/ENTRY_GROUP_ID zu generieren und in einen Cloud Storage-Bucket gs://CLOUD_STORAGE_BUCKET_ID/FOLDER_ID zu schreiben. Bei jeder Ausführung der Pipeline wird im Bucket CLOUD_STORAGE_BUCKET_ID ein neuer Ordner FOLDER_ID erstellt. Der Connector sollte Metadatenimportdateien in diesen Ordner schreiben.

Die Pipelinevorlagen unterstützen PySpark-Connectors. Bei den Vorlagen wird davon ausgegangen, dass der Treiber (mainPythonFileUri) eine lokale Datei im Connector-Image mit dem Namen main.py ist. Sie können die Pipelinevorlagen für andere Szenarien ändern, z. B. für einen Spark-Connector, einen anderen Treiber-URI oder andere Optionen.

So erstellen Sie mit PySpark ein Importelement in der Datei für den Metadatenimport.

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

Hinweis

In diesem Leitfaden wird davon ausgegangen, dass Sie mit Python und PySpark vertraut sind.

Sehen Sie sich die folgenden Informationen an:

Gehen Sie so vor: Alle Ressourcen müssen am selben Google CloudSpeicherort erstellt werden.

  1. Erstellen Sie ein Google Cloud Projekt oder wählen Sie eines aus.

    Rollen, die zum Auswählen oder Erstellen eines Projekts erforderlich sind

    • Projekt auswählen: Für die Auswahl eines Projekts ist keine bestimmte IAM-Rolle erforderlich. Sie können jedes Projekt auswählen, für das Ihnen eine Rolle zugewiesen wurde.
    • Projekt erstellen: Zum Erstellen eines Projekts benötigen Sie die Rolle „Projektersteller“ (roles/resourcemanager.projectCreator), die die Berechtigung resourcemanager.projects.create enthält. Weitere Informationen zum Zuweisen von Rollen

    • So erstellen Sie ein Google Cloud Projekt:

      gcloud projects create PROJECT_ID

      Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud Projekt, das Sie erstellen.

    • Wählen Sie das von Ihnen erstellte Google Cloud Projekt aus:

      gcloud config set project PROJECT_ID

      Ersetzen Sie PROJECT_ID durch den Namen Ihres Projekts in Google Cloud .

  2. Prüfen Sie, ob die Abrechnung für Ihr Google Cloud Projekt aktiviert ist.

  3. Aktivieren Sie die Dataplex-, Dataproc-, Workflows- und Artifact Registry APIs:

    Rollen, die zum Aktivieren von APIs erforderlich sind

    Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Rolle „Inhaber“ (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen 

    gcloud services enable dataplex.googleapis.com dataproc.googleapis.com workflows.googleapis.com artifactregistry.googleapis.com