Vorlage "PostgreSQL für BigQuery"

Die Vorlage "PostgreSQL für BigQuery" ist eine Batchpipeline, die Daten aus einer PostgreSQL-Tabelle in eine vorhandene BigQuery-Tabelle kopiert. Diese Pipeline verwendet JDBC, um eine Verbindung zu PostgreSQL herzustellen. Als zusätzliche Schutzmaßnahme können Sie auch einen Cloud KMS-Schlüssel zusammen mit einem Base64-codierten Nutzernamen, Passwort und Verbindungsstring-Parametern übergeben, die mit dem Cloud KMS-Schlüssel verschlüsselt sind. Weitere Informationen zum Verschlüsseln von Nutzernamen, Passwörtern und Verbindungsstring-Parametern finden Sie unter Cloud KMS API-Verschlüsselungsendpunkt.

Pipelineanforderungen

  • Die BigQuery-Tabelle muss vor der Pipelineausführung vorhanden sein.
  • Die BigQuery-Tabelle muss ein kompatibles Schema haben.
  • Die relationale Datenbank muss über das Subnetz zugänglich sein, in dem Dataflow ausgeführt wird.

Vorlagenparameter

Erforderliche Parameter

  • driverJars : Die durch kommagetrennte Liste der JAR-Dateien des Treibers. (Beispiel: gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar).
  • driverClassName : Der Name der JDBC-Treiberklasse. (Beispiel: com.mysql.jdbc.Driver).
  • connectionURL : Der URL-String für die JDBC-Verbindung. Beispiel: jdbc:mysql://some-host:3306/sampledb. Sie können diesen Wert als String übergeben, der mit einem Cloud KMS-Schlüssel und dann Base64-verschlüsselt ist. Entfernen Sie Leerzeichen aus dem Base64-codierten String. Beachten Sie den Unterschied zwischen einem Oracle-Nicht-RAC-Datenbankverbindungsstring (jdbc:oracle:thin:@some-host:<port>:<sid>) und einem Oracle-RAC-Datenbankverbindungsstring (jdbc:oracle:thin:@//some-host[:<port>]/<service_name>). (Beispiel: jdbc:mysql://some-host:3306/sampledb).
  • outputTable: Der Speicherort der BigQuery-Ausgabetabelle. (Beispiel: <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>).
  • bigQueryLoadingTemporaryDirectory: Das temporäre Verzeichnis für den BigQuery-Ladevorgang (Beispiel: gs://your-bucket/your-files/temp_dir).

Optionale Parameter

  • connectionProperties : Attributstring für die JDBC-Verbindung. Das Format des Strings muss [propertyName=property;]* sein. Weitere Informationen finden Sie unter „Konfigurationsattribute“ (https://dev.mysql.com/doc/connector-j/8.1/en/connector-j-reference-configuration-properties.html) in der MySQL-Dokumentation. Beispiel: unicode=true;characterEncoding=UTF-8).
  • username : Der Nutzername für die JDBC-Verbindung. Sie können diesen Wert als String übergeben, der mit einem Cloud KMS-Schlüssel und dann Base64-verschlüsselt ist. Entfernen Sie Leerzeichen aus dem Base64-codierten String.
  • Passwort : Das Passwort für die JDBC-Verbindung. Sie können diesen Wert als String übergeben, der mit einem Cloud KMS-Schlüssel und dann Base64-verschlüsselt ist. Entfernen Sie Leerzeichen aus dem Base64-codierten String.
  • Abfrage : Die Abfrage, die in der Quelle zur Extraktion der Daten ausgeführt wird. Beachten Sie, dass einige JDBC-SQL- und BigQuery-Typen einige Unterschiede haben, obwohl sie denselben Namen haben. Beachten Sie die folgenden wichtigen SQL -> BigQuery-Typzuordnungen: DATETIME --> ZEITSTEMPEL

Eine Typumwandlung kann erforderlich sein, wenn Ihre Schemas nicht übereinstimmen. (Beispiel: Wählen Sie "*" aus sampledb.sample_table aus).

  • KMSEncryptionKey: Der Cloud KMS-Verschlüsselungsschlüssel zum Entschlüsseln von Nutzernamen, Passwort und Verbindungsstring. Wenn Sie einen Cloud KMS-Schlüssel übergeben, müssen Sie auch den Nutzernamen, das Passwort und den Verbindungsstring verschlüsseln. (Beispiel: projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key).
  • useColumnAlias: Wenn der Wert auf true gesetzt ist, verwendet die Pipeline den Spaltenalias (AS) anstelle des Spaltennamens, um die Zeilen zu BigQuery zuzuordnen. Die Standardeinstellung ist false.
  • isTruncate: Wenn der Wert auf true gesetzt ist, wird die Pipeline vor dem Laden von Daten in BigQuery gekürzt. Die Standardeinstellung ist false, wodurch die Pipeline veranlasst wird, Daten anzuhängen.
  • partitionColumn: Wenn dieser Parameter mit dem Namen des table angegeben wird, der als optionaler Parameter definiert ist, liest JdbcIO die Tabelle parallel, indem mehrere Instanzen der Abfrage in derselben Tabelle (Unterabfrage) mit Bereichen ausgeführt werden. Derzeit werden nur Long-Partitionsspalten unterstützt.
  • table: Die Tabelle, aus der bei Verwendung von Partitionen gelesen werden soll. Dieser Parameter akzeptiert auch eine Unterabfrage in Klammern. (Beispiel: (ID auswählen, Name von Person) als subq).
  • numPartitions : Die Anzahl der Partitionen. Mit der Unter- und Obergrenze bildet dieser Wert Partitionsschritte für generierte WHERE-Anweisungsausdrücke, die zum gleichmäßigen Aufteilen der Partitionsspalte verwendet werden. Wenn die Eingabe kleiner als 1 ist, wird die Zahl auf 1 gesetzt.
  • lowerBound: Die Untergrenze, die im Partitionsschema verwendet werden soll. Wenn nicht angegeben, wird dieser Wert von Apache Beam automatisch für die unterstützten Typen abgeleitet.
  • upperBound: Die Obergrenze, die im Partitionsschema verwendet werden soll. Wenn nicht angegeben, wird dieser Wert von Apache Beam automatisch für die unterstützten Typen abgeleitet.
  • fetchSize : Die Anzahl der Zeilen, die jeweils aus der Datenbank abgerufen werden sollen. Wird nicht für partitionierte Lesevorgänge verwendet. Die Standardeinstellung ist 50000.
  • createDisposition: Die zu verwendende BigQuery-CreateDisposition-Einstellung. Beispiel: CREATE_IF_NEEDEDoder CREATE_NEVER Die Standardeinstellung ist CREATE_NEVER.
  • bigQuerySchemaPath : Der Cloud Storage-Pfad für das BigQuery-JSON-Schema. Wenn createDisposition auf CREATE_IF_NEEDED gesetzt ist, muss dieser Parameter angegeben werden. (Beispiel: gs://your-bucket/your-schema.json).
  • disabledAlgorithms : Durch Kommas getrennte Algorithmen zum Deaktivieren. Wenn dieser Wert auf „Keine“ gesetzt ist, wird kein Algorithmus deaktiviert. Verwenden Sie diesen Parameter mit Vorsicht, da die standardmäßig deaktivierten Algorithmen Sicherheitslücken oder Leistungsprobleme haben können. (Beispiel: SSLv3, RC4).
  • extraFilesToStage : Durch Kommas getrennte Cloud Storage-Pfade oder Secret Manager-Secrets für Dateien, die im Worker bereitgestellt werden sollen. Diese Dateien werden im Verzeichnis /extra_files in jedem Worker gespeichert. (Beispiel: gs://
  • defaultLogLevel: Legen Sie die Protokollebene in den Workern fest. Unterstützte Optionen sind OFF, ERROR, WARN, INFO, DEBUG, TRACE. Die Standardeinstellung ist INFO.
  • useStorageWriteApi: Wenn true, verwendet die Pipeline die BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api). Der Standardwert ist false. Weitere Informationen finden Sie unter „Storage Write API verwenden“ (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
  • useStorageWriteApiAtLeastOnce: Gibt bei Verwendung der Storage Write API die Schreibsemantik an. Wenn Sie die "Mindestens einmal"-Semantik verwenden möchten (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), legen Sie diesen Parameter auf true fest. Wenn Sie die "Genau einmal"-Semantik verwenden möchten, legen Sie den Parameter auf false fest. Dieser Parameter gilt nur, wenn useStorageWriteApi true ist. Der Standardwert ist false.

Führen Sie die Vorlage aus.

Console

  1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
    2. Zur Seite "Job aus Vorlage erstellen“
  2. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
  3. Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.

    Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.

  4. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the PostgreSQL to BigQuery templateaus.
  5. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
  6. Klicken Sie auf Job ausführen.

gcloud

Führen Sie die Vorlage in der Shell oder im Terminal aus:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/PostgreSQL_to_BigQuery \
    --parameters \
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=PROJECT_ID:DATASET.TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

Ersetzen Sie dabei Folgendes:

  • JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
  • VERSION: Die Version der Vorlage, die Sie verwenden möchten

    Sie können die folgenden Werte verwenden:

    • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
    • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  • REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
  • JDBC_CONNECTION_URL: Die JDBC-Verbindungs-URL
  • SOURCE_SQL_QUERY: die SQL-Abfrage, die in der Quelldatenbank ausgeführt werden soll
  • DATASET: Ihr BigQuery-Dataset
  • TABLE_NAME: Ihr BigQuery-Tabellenname
  • PATH_TO_TEMP_DIR_ON_GCS: Der Cloud Storage-Pfad zum temporären Verzeichnis
  • CONNECTION_PROPERTIES: die JDBC-Verbindungsattribute, falls erforderlich
  • CONNECTION_USERNAME: Der Nutzername der JDBC-Verbindung
  • CONNECTION_PASSWORD: Das JDBC-Verbindungspasswort
  • KMS_ENCRYPTION_KEY: der Cloud KMS-Verschlüsselungsschlüssel

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launchParameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/PostgreSQL_to_BigQuery"
    "parameters": {
      "connectionURL": "JDBC_CONNECTION_URL",
      "query": "SOURCE_SQL_QUERY",
      "outputTable": "PROJECT_ID:DATASET.TABLE_NAME",
      "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
      "connectionProperties": "CONNECTION_PROPERTIES",
      "username": "CONNECTION_USERNAME",
      "password": "CONNECTION_PASSWORD",
      "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
    },
    "environment": { "zone": "us-central1-f" }
  }
}

Ersetzen Sie dabei Folgendes:

  • PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
  • JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
  • VERSION: Die Version der Vorlage, die Sie verwenden möchten

    Sie können die folgenden Werte verwenden:

    • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
    • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  • LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
  • JDBC_CONNECTION_URL: Die JDBC-Verbindungs-URL
  • SOURCE_SQL_QUERY: die SQL-Abfrage, die in der Quelldatenbank ausgeführt werden soll
  • DATASET: Ihr BigQuery-Dataset
  • TABLE_NAME: Ihr BigQuery-Tabellenname
  • PATH_TO_TEMP_DIR_ON_GCS: Der Cloud Storage-Pfad zum temporären Verzeichnis
  • CONNECTION_PROPERTIES: die JDBC-Verbindungsattribute, falls erforderlich
  • CONNECTION_USERNAME: Der Nutzername der JDBC-Verbindung
  • CONNECTION_PASSWORD: Das JDBC-Verbindungspasswort
  • KMS_ENCRYPTION_KEY: der Cloud KMS-Verschlüsselungsschlüssel

Nächste Schritte