Streams verwalten

Überblick

In diesem Abschnitt erfahren Sie, wie Sie die Datastream API für folgende Aufgaben verwenden:

  • Streams erstellen
  • Informationen zu Streams und Streamobjekten abrufen
  • Streams aktualisieren, indem Sie sie starten, pausieren, fortsetzen und ändern sowie Backfill für Streamobjekte initiieren und beenden
  • Dauerhaft fehlgeschlagene Streams wiederherstellen
  • Streaming großer Objekte für Oracle-Streams aktivieren
  • Streams löschen

Sie können die Datastream API auf zwei Arten verwenden. Sie können REST API-Aufrufe ausführen oder die Google Cloud CLI (CLI) verwenden.

Allgemeine Informationen zur Verwendung von gcloud zum Verwalten von Datastream-Streams finden Sie unter gcloud-Datastream-Streams.

Stream erstellen

In diesem Abschnitt erfahren Sie, wie Sie einen Stream zum Übertragen von Daten aus Ihrer Quelle an ein Ziel erstellen. Die folgenden Beispiele sind nicht umfassend, sondern heben spezifische Funktionen von Datastream hervor. Verwenden Sie diese Beispiele zusammen mit der Referenzdokumentation zur Datastream API für Ihren spezifischen Anwendungsfall.

In diesem Abschnitt werden die folgenden Anwendungsfälle behandelt:

Beispiel 1: Bestimmte Objekte in BigQuery streamen

In diesem Beispiel lernen Sie Folgendes:

  • Von MySQL zu BigQuery streamen
  • Eine Reihe von Objekten in den Stream aufnehmen
  • Definieren Sie den Schreibmodus für den Stream als „Nur Anfügung“.
  • Backfill für alle im Stream enthaltenen Objekte ausführen

Das folgende Beispiel zeigt eine Anfrage zum Abrufen aller Tabellen aus schema1 und zwei bestimmten Tabellen aus schema2: tableA und tableC. Die Ereignisse werden in ein Dataset in BigQuery geschrieben.

Die Anfrage enthält den Parameter customerManagedEncryptionKey nicht. Daher wird das interne Schlüsselverwaltungssystem von Google Cloud anstelle des CMEK zum Verschlüsseln Ihrer Daten verwendet.

Für den Parameter backfillAll, der mit dem Ausführen des Backfills (oder Snapshots) für den Verlauf verknüpft ist, ist ein leeres Wörterbuch ({}) festgelegt. Das bedeutet, dass Datastream die Verlaufsdaten aus allen im Stream enthaltenen Tabellen füllt.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlCdcStream
{
  "displayName": "MySQL CDC to BigQuery",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "schema1" },
          {
            "database": "schema2",
            "mysqlTables": [
              {
                "table": "tableA",
                "table": "tableC"
              }
            ]
          }
        ]
      },
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "dataFreshness": "900s"
    }
  },
  "backfillAll": {}
}

gcloud

Weitere Informationen zum Erstellen eines Streams mit gcloud finden Sie in der Google Cloud SDK-Dokumentation.

Beispiel 2: Bestimmte Objekte aus einem Stream mit einer PostgreSQL-Quelle ausschließen

In diesem Beispiel lernen Sie Folgendes:

  • Von PostgreSQL zu BigQuery streamen
  • Objekte aus dem Stream ausschließen
  • Objekte aus dem Backfill ausschließen

Der folgende Code zeigt eine Anfrage zum Erstellen eines Streams, mit dem Daten aus einer PostgreSQL-Quelldatenbank an BigQuery übertragen werden. Wenn Sie einen Stream aus einer PostgreSQL-Quelldatenbank erstellen, müssen Sie in der Anfrage zwei zusätzliche, PostgreSQL-spezifische Felder angeben:

  • replicationSlot: Ein Replikationsslot ist eine Voraussetzung zum Konfigurieren einer PostgreSQL-Datenbank für die Replikation. Sie müssen für jeden Stream einen Replikationsslot erstellen.
  • publication: Eine Publikation ist eine Gruppe von Tabellen, aus denen Sie Änderungen replizieren möchten. Der Publikationsname muss in der Datenbank vorhanden sein, bevor ein Stream gestartet wird. Die Publikation muss mindestens die Tabellen enthalten, die in der includeObjects-Liste des Streams angegeben sind.

Der backfillAll-Parameter, der mit dem bisherigen Backfill (oder Snapshot) verknüpft ist, ist so eingestellt, dass eine Tabelle ausgeschlossen wird.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myPostgresStream
{
  "displayName": "PostgreSQL to BigQueryCloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/postgresCp",
    "postgresqlSourceConfig": {
      "replicationSlot": "replicationSlot1",
      "publication": "publicationA",
      "includeObjects": {
        "postgresqlSchemas": {
          "schema": "schema1"
        }
      },
      "excludeObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
        "postgresqlTables": [
          {
            "table": "tableA",
            "postgresqlColumns": [
              { "column": "column5" }
              ]
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
           "location": "us",
           "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "backfillAll": {
    "postgresqlExcludedObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
            "postgresqlTables": [
              { "table": "tableA" }
            ]
          }
        ]
      }
    }
  }

gcloud

Weitere Informationen zum Erstellen eines Streams mit gcloud finden Sie in der Google Cloud SDK-Dokumentation.

Beispiel 3: Schreibmodus „Nur Anfügungen“ für einen Stream angeben

Beim Streaming nach BigQuery können Sie den Schreibmodus definieren: merge oder appendOnly. Weitere Informationen finden Sie unter Schreibmodus konfigurieren.

Wenn Sie in der Anfrage keinen Schreibmodus angeben, um einen Stream zu erstellen, wird der Standardmodus merge verwendet.

In der folgenden Anfrage wird gezeigt, wie Sie den appendOnly-Modus beim Erstellen eines MySQL-zu-BigQuery-Streams definieren.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=appendOnlyStream
{
  "displayName": "My append-only stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

Weitere Informationen zum Erstellen eines Streams mit gcloud finden Sie in der Google Cloud SDK-Dokumentation.

Beispiel 4: An ein Cloud Storage-Ziel streamen

In diesem Beispiel lernen Sie Folgendes:

  • Von Oracle zu Cloud Storage streamen
  • Gruppe von Objekten definieren, die in den Stream aufgenommen werden sollen
  • CMEK zum Verschlüsseln ruhender Daten definieren

In der folgenden Anfrage wird gezeigt, wie Sie einen Stream erstellen, der die Ereignisse in einen Bucket in Cloud Storage schreibt.

In dieser Beispielanfrage werden die Ereignisse im JSON-Ausgabeformat geschrieben und alle 100 MB bzw. 30 Sekunden wird eine neue Datei erstellt. Dabei werden die Standardwerte von 50 MB und 60 Sekunden überschrieben.

Für das JSON-Format können Sie:

  • Fügen Sie eine einheitliche Dockerfile-Datei in den Pfad ein. Dadurch schreibt Datastream zwei Dateien in Cloud Storage: eine JSON-Datendatei und eine Avro-Schemadatei. Die Schemadatei hat denselben Namen wie die Datendatei, aber mit der Erweiterung .schema.

  • gzip-Komprimierung aktivieren, damit Datastream die Dateien komprimiert, die in Cloud Storage geschrieben werden.

Mithilfe des Parameters backfillNone gibt die Anfrage an, dass nur laufende Änderungen ohne Backfill in das Ziel gestreamt werden.

Die Anfrage gibt den vom Kunden verwalteten Verschlüsselungsschlüsselparameter an, mit dem Sie die Schlüssel steuern können, die zum Verschlüsseln inaktiver Daten in einem Google Cloud-Projekt verwendet werden. Der Parameter bezieht sich auf den CMEK, mit dem Datastream Daten verschlüsselt, die von der Quelle zum Ziel gestreamt werden. Außerdem wird der Schlüsselbund für Ihren CMEK angegeben.

Weitere Informationen zu Schlüsselbunden finden Sie unter Cloud KMS-Ressourcen. Weitere Informationen zum Schutz Ihrer Daten mit Verschlüsselungsschlüsseln finden Sie unter Cloud Key Management Service (KMS).

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleCdcStream
{
  "displayName": "Oracle CDC to Cloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/
    connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "GcsBucketCp",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "jsonFileFormat": {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      },
      "fileRotationMb": 100,
      "fileRotationInterval": 30
    }
  },
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillNone": {}
}

gcloud

Weitere Informationen zum Erstellen eines Streams mit gcloud finden Sie in der Google Cloud SDK-Dokumentation.

Definition eines Streams validieren

Bevor Sie einen Stream erstellen, können Sie seine Definition überprüfen. So sorgen Sie dafür, dass alle Validierungsprüfungen bestanden werden und der Stream nach der Erstellung erfolgreich ausgeführt wird.

Dabei wird Folgendes geprüft:

  • Ob die Quelle ordnungsgemäß konfiguriert ist, damit Datastream Daten von ihr streamen kann.
  • Ob er sowohl mit der Quelle als auch mit dem Ziel verbunden werden kann.
  • Die End-to-End-Konfiguration des Streams.

Fügen Sie der URL vor dem Text Ihrer Anfrage &validate_only=true hinzu, um einen Stream zu validieren:

POST "https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID&validate_only=true"

Nachdem Sie diese Anfrage gestellt haben, werden die Validierungsprüfungen angezeigt, die Datastream für Ihre Quelle und Ihr Ziel ausführt, und ob die Prüfungen bestanden werden oder fehlschlagen. Bei nicht bestandenen Validierungsprüfungen werden Informationen dazu angezeigt, warum die Prüfung fehlgeschlagen ist und wie sie das Problem beheben können.

Angenommen, Sie haben einen vom Kunden verwalteten Verschlüsselungsschlüssel (CMEK), mit dem Datastream Daten verschlüsseln soll, die von der Quelle zum Ziel gestreamt werden. Bei der Validierung des Streams prüft Datastream, ob der Schlüssel vorhanden ist und ob Datastream die Berechtigungen zur Verwendung des Schlüssels hat. Wenn eine dieser Bedingungen nicht erfüllt ist, wird bei der Validierung des Streams die folgende Fehlermeldung zurückgegeben:

CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS

Prüfen Sie, ob der von Ihnen angegebene Schlüssel existiert und das Datastream-Dienstkonto die Berechtigung cloudkms.cryptoKeys.get für den Schlüssel hat, um dieses Problem zu beheben.

Nachdem Sie die entsprechenden Korrekturen vorgenommen haben, stellen Sie die Anfrage noch einmal, um sicherzustellen, dass alle Validierungsprüfungen bestanden wurden. Im obigen Beispiel gibt die CMEK_VALIDATE_PERMISSIONS-Prüfung keine Fehlermeldung mehr zurück, hat aber den Status PASSED.

Informationen zu einem Stream abrufen

Der folgende Code zeigt eine Anfrage zum Abrufen von Informationen zu einem Stream. Zu diesen Daten gehören:

  • Der Name des Streams (eindeutige Kennung)
  • Ein nutzerfreundlicher Name für den Stream (Anzeigename)
  • Zeitstempel, der angibt, wann der Stream erstellt und zuletzt aktualisiert wurde
  • Informationen zu den Quell- und Zielverbindungsprofilen, die mit dem Stream verknüpft sind
  • Der Status des Streams

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams/STREAM_ID

Die Antwort wird so angezeigt:

{
  "name": "myOracleCdcStream",
  "displayName": "Oracle CDC to Cloud Storage",
  "createTime": "2019-12-15T15:01:23.045123456Z",
  "updateTime": "2019-12-15T15:01:23.045123456Z",
  "sourceConfig": {
    "sourceConnectionProfileName": "myOracleDb",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          },
          {
            "schema": "schema3",
            "oracleTables": [
              { "table": "tableA" },
              { "table": "tableC" }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "myGcsBucket",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "avroFileFormat": {},
      "fileRotationMb": 100,
      "fileRotationInterval": 60
    }
  },
  "state": "RUNNING"
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillAll": {}
}

gcloud

Weitere Informationen dazu, wie Sie mithilfe von gcloud Informationen zu Ihrem Stream abrufen, finden Sie hier.

Streams auflisten

Der folgende Code zeigt eine Anfrage zum Abrufen einer Liste aller Streams im angegebenen Projekt und am angegebenen Standort.

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams

gcloud

Weitere Informationen dazu, wie Sie mithilfe von gcloud Informationen zu allen Ihren Streams abrufen können, finden Sie hier.

Objekte eines Streams auflisten

Der folgende Code zeigt eine Anfrage zum Abrufen von Informationen zu allen Objekten eines Streams.

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects

gcloud

Weitere Informationen zur Verwendung von gcloud zum Abrufen von Informationen zu allen Objekten Ihres Streams finden Sie hier.

Die Liste der zurückgegebenen Objekte sieht in etwa so aus:

REST

{
  "streamObjects": [
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object1",
      "displayName": "employees.salaries",
      "backfillJob": {
        "state": "ACTIVE",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T12:12:26.344878Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "salaries"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object2",
      "displayName": "contractors.hours",
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "contractors",
          "table": "hours"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object3",
      "displayName": "employees.departments",
      "backfillJob": {
        "state": "COMPLETED",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T11:26:12.869880Z",
        "lastEndTime": "2021-10-18T11:26:28.405653Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "departments"
        }
      }
    }
  ]
}

gcloud

Weitere Informationen zur Verwendung von gcloud zum Auflisten der Objekte eines Streams finden Sie hier.

Stream starten

Der folgende Code zeigt eine Anfrage zum Starten eines Streams.

Bei Verwendung des Parameters updateMask in der Anfrage müssen nur die von Ihnen angegebenen Felder im Text der Anfrage enthalten sein. Ändern Sie den Wert im Feld state von CREATED in RUNNING, um einen Stream zu starten.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

Weitere Informationen zur Verwendung von gcloud zum Starten deines Streams findest du hier.

Stream pausieren

Der folgende Code zeigt eine Anfrage zum Anhalten eines ausgeführten Streams.

In diesem Beispiel ist für den Parameter updateMask das Feld state angegeben. Wenn Sie den Stream pausieren, ändern Sie seinen Status von RUNNING in PAUSED.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "PAUSED"
}

gcloud

Weitere Informationen zum Pausieren deines Streams mit gcloud findest du hier.

Stream fortsetzen

Der folgende Code zeigt eine Anfrage zum Fortsetzen eines angehaltenen Streams.

In diesem Beispiel ist für den Parameter updateMask das Feld state angegeben. Wenn Sie den Stream fortsetzen, ändern Sie seinen Status vonPAUSED in RUNNING.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

Weitere Informationen dazu, wie du deinen Stream über gcloud fortsetzen kannst, findest du hier.

Stream wiederherstellen

Mit der Methode RunStream können Sie einen dauerhaft fehlerhaften Stream für eine MySQL-, Oracle- oder PostgreSQL-Quelle wiederherstellen. Jeder Quelldatenbanktyp hat eine eigene Definition dafür, welche Streamwiederherstellungsvorgänge möglich sind. Weitere Informationen finden Sie unter Stream wiederherstellen.

Stream für eine MySQL- oder Oracle-Quelle wiederherstellen

Die folgenden Codebeispiele zeigen Anfragen zum Wiederherstellen eines Streams für eine MySQL- oder Oracle-Quelle aus verschiedenen Logdateipositionen:

REST

Stream von der aktuellen Position wiederherstellen Dies ist die Standardoption:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

So stellen Sie einen Stream ab der nächsten verfügbaren Position wieder her:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

So stellen Sie einen Stream ab der letzten Position wieder her:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "mostRecentStartPosition": {}
  }
}

Stellen Sie einen Stream von einer bestimmten Position aus wieder her (MySQL):

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

Ersetzen Sie Folgendes:

  • NAME_OF_THE_LOG_FILE: Der Name der Logdatei, aus der Sie den Stream wiederherstellen möchten.
  • POSITION: Die Position in der Logdatei, von der aus Sie den Stream wiederherstellen möchten. Wenn Sie den Wert nicht angeben, stellt Datastream den Stream aus dem Header der Datei wieder her.

Beispiel:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 4
      }
    }
  }
}

Stream von einer bestimmten Position wiederherstellen (Oracle):

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}
Ersetzen Sie scn durch die Systemänderungsnummer (SCN) in der Redo-Logdatei, aus der Sie den Stream wiederherstellen möchten. Dies ist ein Pflichtfeld.

Beispiel:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 234234
      }
    }
  }
}

Weitere Informationen zu verfügbaren Wiederherstellungsoptionen finden Sie unter Stream wiederherstellen.

gcloud

Das Wiederherstellen eines Streams mit gcloud wird nicht unterstützt.

Stream für eine PostgreSQL-Quelle wiederherstellen

Das folgende Codebeispiel zeigt eine Anfrage zum Wiederherstellen eines Streams für eine PostgreSQL-Quelle. Während der Wiederherstellung beginnt der Stream mit dem Lesen der ersten Logsequenznummer (LSN) im Replikationsslot, der für den Stream konfiguriert ist.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

Beispiel:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run

Wenn Sie den Replikationsslot ändern möchten, aktualisieren Sie zuerst den Stream mit dem neuen Namen des Replikationsslots:

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.postgresqlSourceConfig.replicationSlot
{
  "sourceConfig": {
    "postgresqlSourceConfig": {
      "replicationSlot": "NEW_REPLICATION_SLOT_NAME"
    }
  }
}

gcloud

Das Wiederherstellen eines Streams mit gcloud wird nicht unterstützt.

Stream ab einer bestimmten Position starten oder fortsetzen

Sie können einen Stream starten oder einen pausierten Stream von einer bestimmten Position für MySQL- und Oracle-Quellen fortsetzen. Dies kann nützlich sein, wenn Sie einen Backfill mit einem externen Tool ausführen oder das CDC von einer von Ihnen angegebenen Position aus starten möchten. Für eine MySQL-Quelle müssen Sie eine binlog-Position und für eine Oracle-Quelle eine System Change Number (SCN) in der Redo-Logdatei angeben.

Der folgende Code zeigt eine Anfrage zum Starten oder Fortsetzen eines bereits erstellten Streams ab einer bestimmten Position.

Starten oder fortsetzen Sie einen Stream von einer bestimmten binlog-Position aus (MySQL):

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

Ersetzen Sie Folgendes:

  • NAME_OF_THE_LOG_FILE: Der Name der Logdatei, von der Sie den Stream starten möchten.
  • POSITION: Die Position in der Logdatei, von der aus Sie den Stream starten möchten. Wenn Sie den Wert nicht angeben, beginnt Datastream mit dem Lesen des Headers der Datei.

Beispiel:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 2
      }
    }
  }
}

gcloud

Das Starten oder Fortsetzen eines Streams ab einer bestimmten Position mit gcloud wird nicht unterstützt. Informationen zur Verwendung von gcloud zum Starten oder Fortsetzen eines Streams finden Sie in der Cloud SDK-Dokumentation.

Starten oder fortsetzen Sie einen Stream ab einer bestimmten Systemänderungsnummer in der Redo-Logdatei (Oracle):

REST

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}
Ersetzen Sie scn durch die Systemänderungsnummer (SCN) in der Redo-Logdatei, über die Sie den Stream starten möchten. Dies ist ein Pflichtfeld.

Beispiel:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 123123
      }
    }
  }
}

gcloud

Das Starten oder Fortsetzen eines Streams ab einer bestimmten Position mit gcloud wird nicht unterstützt. Informationen zur Verwendung von gcloud zum Starten eines Streams finden Sie in der Cloud SDK-Dokumentation.

Stream ändern

Der folgende Code zeigt eine Anfrage zum Aktualisieren der Dateirotationskonfiguration eines Streams, um die Datei alle 75 MB oder 45 Sekunden zu rotieren.

In diesem Beispiel werden für den Parameter updateMask die Felder fileRotationMb und fileRotationInterval angegeben, die durch die Flags destinationConfig.gcsDestinationConfig.fileRotationMb bzw. destinationConfig.gcsDestinationConfig.fileRotationInterval dargestellt sind.

REST

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
fileRotationMb,destinationConfig.gcsDestinationConfig.fileRotationInterval
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "fileRotationMb": 75,
      "fileRotationInterval": 45
    }
  }
}

Der folgende Code zeigt eine Anfrage zum Einfügen einer Unified Types-Schemadatei in den Pfad von Dateien, die Datastream in Cloud Storage schreibt. Daher schreibt Datastream zwei Dateien: eine JSON-Datendatei und eine Avro-Schemadatei.

In diesem Beispiel ist das Feld jsonFileFormat, dargestellt durch das Flag destinationConfig.gcsDestinationConfig.jsonFileFormat.

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
jsonFileFormat
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "jsonFileFormat" {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      }  
    }
  }
}

Der folgende Code zeigt eine Anfrage an Datastream, um vorhandene Daten sowie laufende Änderungen an den Daten aus der Quelldatenbank in das Ziel zu replizieren.

Im Abschnitt oracleExcludedObjects des Codes werden die Tabellen und Schemas aufgeführt, die nicht für das Backfill in das Ziel verwendet werden dürfen.

Für dieses Beispiel werden alle Tabellen und Schemas außer tableA in schema3 ausgefüllt.

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=backfillAll
{
  "backfillAll": {
    "oracleExcludedObjects": {
      "oracleSchemas": [
        {
          "schema": "schema3",
          "oracleTables": [
            { "table": "tableA" }
          ]
        }
      ]
    }
  }
}

gcloud

Weitere Informationen zur Verwendung von gcloud zum Ändern Ihres Streams finden Sie hier.

Backfill für ein Objekt eines Streams initiieren

Ein Stream in Datastream kann ein Backfill auf Verlaufsdaten ausführen und laufende Änderungen an ein Ziel streamen. Laufende Änderungen werden immer von einer Quelle an ein Ziel gestreamt. Sie können jedoch festlegen, ob Verlaufsdaten gestreamt werden sollen.

Wenn Sie möchten, dass Verlaufsdaten von der Quelle an das Ziel gestreamt werden, verwenden Sie den Parameter backfillAll.

Mit Datastream können Sie auch Verlaufsdaten nur für bestimmte Datenbanktabellen streamen. Verwenden Sie dazu den Parameter backfillAll und schließen Sie die Tabellen aus, für die Sie keine Verlaufsdaten abrufen möchten.

Wenn nur laufende Änderungen an das Ziel gestreamt werden sollen, verwenden Sie den Parameter backfillNone. Wenn Sie möchten, dass Datastream einen Snapshot aller vorhandenen Daten von der Quelle zum Ziel streamt, müssen Sie den Backfill für die Objekte, die diese Daten enthalten, manuell initiieren.

Ein weiterer Grund zum Initiieren eines Backfills für ein Objekt ist, wenn die Daten zwischen der Quelle und dem Ziel nicht synchron sind. Beispielsweise kann ein Nutzer versehentlich Daten im Ziel löschen und die Daten gehen nun verloren. In diesem Fall dient das Initiieren des Backfills für das Objekt als "Rücksetzungsmechanismus", da alle Daten auf einmal an das Ziel gestreamt werden. Folglich werden die Daten zwischen der Quelle und dem Ziel synchronisiert.

Bevor Sie den Backfill für ein Objekt eines Streams initiieren können, müssen Sie Informationen zum Objekt abrufen.

Jedes Objekt hat einen OBJECT_ID, der das Objekt eindeutig identifiziert. Mit OBJECT_ID starten Sie den Backfill für den Stream.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:startBackfillJob

gcloud

Weitere Informationen zur Verwendung von gcloud zum Initiieren des Backfills für ein Objekt Ihres Streams finden Sie in der Google Cloud SDK-Dokumentation.

Backfill für ein Objekt eines Streams beenden

Nach dem Initialisieren eines Backfills für ein Objekt eines Streams können Sie den Backfill für das Objekt beenden. Wenn ein Nutzer beispielsweise ein Datenbankschema ändert, können das Schema oder die Daten beschädigt sein. Da dieses Schema oder die Daten nicht an das Ziel gestreamt werden sollen, beenden Sie den Backfill für das Objekt.

Sie können den Backfill für ein Objekt auch zum Load-Balancing beenden. Datastream kann mehrere Backfills parallel ausführen. Dies kann eine zusätzliche Last für die Quelle bedeuten. Wenn die Last erheblich ist, beenden Sie den Backfill für jedes Objekt und initiieren dann den Backfill für die Objekte nacheinander.

Bevor Sie den Backfill für ein Objekt eines Streams beenden können, müssen Sie eine Anfrage zum Abrufen von Informationen zu allen Objekten eines Streams stellen. Jedes zurückgegebene Objekt hat einen OBJECT_ID, der das Objekt eindeutig identifiziert. Mit dem OBJECT_ID können Sie den Backfill für den Stream beenden.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:stopBackfillJob

gcloud

Weitere Informationen zur Verwendung von gcloud zum Beenden des Backfills für ein Objekt Ihres Streams finden Sie hier.

Anzahl der maximalen gleichzeitigen CDC-Aufgaben ändern

Der folgende Code zeigt, wie Sie die maximale Anzahl von CDC-Aufgaben (Parallel Change Data Capture) für einen MySQL-Stream auf 7 festlegen.

In diesem Beispiel ist für den Parameter updateMask das Feld maxConcurrentCdcTasks angegeben. Wenn Sie den Wert auf 7 festlegen, ändern Sie die Anzahl der maximalen gleichzeitigen CDC-Aufgaben vom vorherigen Wert in 7. Sie können Werte von 0 bis 50 (einschließlich) verwenden. Wenn Sie den Wert nicht oder als 0 definieren, wird für den Stream der Standardwert von 5 Aufgaben verwendet.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentCdcTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentCdcTasks": "7"
      }
    }  
}

gcloud

Weitere Informationen zur Verwendung von gcloud finden Sie hier.

Maximale Anzahl gleichzeitiger Backfill-Aufgaben ändern

Der folgende Code zeigt, wie die Anzahl der maximalen gleichzeitigen Backfill-Aufgaben für einen MySQL-Stream auf 25 festgelegt wird.

In diesem Beispiel ist für den Parameter updateMask das Feld maxConcurrentBackfillTasks angegeben. Wenn Sie den Wert auf 25 setzen, ändern Sie die Anzahl der maximalen gleichzeitigen Backfill-Aufgaben vom vorherigen Wert in 25. Sie können Werte von 0 bis 50 (einschließlich) verwenden. Wenn Sie den Wert nicht oder als 0 definieren, wird für den Stream der Standardwert von 16 Aufgaben verwendet.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/
streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentBackfillTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentBackfillTasks": "25"
      }
    }  
}

gcloud

Weitere Informationen zur Verwendung von gcloud finden Sie hier.

Streaming großer Objekte für Oracle-Quellen aktivieren

Sie können das Streaming großer Objekte, z. B. binäre große Objekte (BLOB), große zeichengroße Objekte (CLOB) und große Objekte mit nationalen Zeichen (NCLOB), für Streams mit Oracle-Quellen aktivieren. Mit dem Flag streamLargeObjects können Sie große Objekte sowohl in neue als auch in vorhandene Streams einbeziehen. Das Flag wird auf Streamebene festgelegt. Sie müssen die Spalten großer Objektdatentypen nicht angeben.

Das folgende Beispiel zeigt, wie Sie einen Stream zum Streamen großer Objekte erstellen.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleLobStream
{
  "displayName": "Oracle LOB stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1",
            "oracleTables": [
              {
                "table": "tableA",
                "oracleColumns": [
                  {
                    "column": "column1,column2"
                  }
                ]
              }
            ]
          }
        ]
      },
      "excludeObjects": {},
      "streamLargeObjects": {}
    }
  }
}

gcloud

Weitere Informationen zur Verwendung von gcloud zum Aktualisieren eines Streams finden Sie in der Google Cloud SDK-Dokumentation.

Stream löschen

Der folgende Code zeigt eine Anfrage zum Löschen eines Streams.

REST

DELETE https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID

gcloud

Weitere Informationen zur Verwendung von gcloud zum Löschen Ihres Streams finden Sie hier.

Zurück
Verbindungsprofile mithilfe der API verwalten
Weiter
Probleme mit einem Stream beheben