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 den Backfill für Streamobjekte initiieren und beenden
  • Endgültig 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 erstellen, um Daten von der Quelle an ein Ziel zu übertragen. Die folgenden Beispiele sind nicht umfassend, sondern heben vielmehr bestimmte Features von Datastream hervor. Für Ihren speziellen Anwendungsfall verwenden Sie diese Beispiele zusammen mit der Referenzdokumentation zur Datastream API.

In diesem Abschnitt werden die folgenden Anwendungsfälle behandelt:

Beispiel 1: Bestimmte Objekte zu BigQuery streamen

In diesem Beispiel erfahren Sie, wie Sie:

  • Von MySQL zu BigQuery streamen
  • Eine Reihe von Objekten in den Stream einbeziehen
  • Backfill für alle im Stream enthaltenen Objekte

Im Folgenden sehen Sie eine Anfrage zum Abrufen aller Tabellen aus schema1 und zwei bestimmte 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 eines CMEK zum Verschlüsseln Ihrer Daten verwendet.

Der Parameter backfillAll, der mit der Durchführung des bisherigen Backfills (oder Snapshots) verknüpft ist, ist auf ein leeres Wörterbuch ({}) festgelegt. Das bedeutet, dass Datastream die Verlaufsdaten aus allen im Stream enthaltenen Tabellen auffü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": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "backfillAll": {}
}

gcloud

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

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

In diesem Beispiel erfahren Sie, wie Sie:

  • 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, der zum Übertragen von Daten aus einer PostgreSQL-Quelldatenbank zu BigQuery verwendet wird. Wenn Sie einen Stream aus einer PostgreSQL-Quelldatenbank erstellen, müssen Sie in Ihrer 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 Name der Publikation muss vor dem Starten eines Streams in der Datenbank vorhanden sein. Die Publikation muss mindestens die Tabellen enthalten, die in der Liste includeObjects des Streams angegeben sind.

Der Parameter backfillAll, der mit der Ausführung des bisherigen Backfills (oder Snapshots) verknüpft ist, ist so festgelegt, 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 zur Verwendung von gcloud zum Erstellen eines Streams finden Sie in der Google Cloud SDK-Dokumentation.

Beispiel 3: An ein Cloud Storage-Ziel streamen

In diesem Beispiel erfahren Sie, wie Sie:

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

Die folgende Anfrage zeigt, 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 oder 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 in den Pfad eine SDF-Datei mit einheitlichen Typen 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.

Mit dem Parameter backfillNone gibt die Anfrage an, dass nur laufende Änderungen ohne Backfill an das Ziel gestreamt werden.

Die Anfrage gibt den vom Kunden verwalteten Verschlüsselungsschlüsselparameter an, mit dem Sie die Schlüssel zum Verschlüsseln ruhender Daten in einem Google Cloud-Projekt steuern können. 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 zur Verwendung von gcloud zum Erstellen eines Streams finden Sie in der Google Cloud SDK-Dokumentation.

Definition eines Streams validieren

Bevor Sie einen Stream erstellen, können Sie dessen Definition validieren. Auf diese Weise können Sie dafür sorgen, dass alle Validierungsprüfungen bestanden werden und der Stream nach dem Erstellen 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.

Zum Validieren eines Streams fügen Sie der URL vor dem Text Ihrer Anfrage &validate_only=true hinzu:

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. Im Rahmen der Validierung des Streams prüft Datastream, ob der Schlüssel vorhanden ist und ob Datastream Berechtigungen zur Verwendung des Schlüssels hat. Wenn eine dieser Bedingungen nicht erfüllt ist, wird beim Validieren des Streams die folgende Fehlermeldung zurückgegeben:

CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS

Prüfen Sie zum Beheben dieses Problems, ob der von Ihnen angegebene Schlüssel vorhanden ist und ob das Datastream-Dienstkonto die Berechtigung cloudkms.cryptoKeys.get für den Schlüssel hat.

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 Erstellung und letzten Aktualisierung des Streams
  • 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 zur Verwendung von gcloud zum Abrufen von Informationen zu Ihrem Stream finden Sie hier.

Streams auflisten

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

REST

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

gcloud

Weitere Informationen zur Verwendung von gcloud zum Abrufen von Informationen zu allen Ihren Streams 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 könnte etwa so aussehen:

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 von Objekten eines Streams finden Sie hier.

Stream starten

Der folgende Code stellt eine Anfrage zum Starten eines Streams dar.

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 Ihres Streams finden Sie hier.

Stream ab einer bestimmten Position starten

Sie können einen Stream ab einer bestimmten Position für MySQL- und Oracle-Quellen starten, z. B. wenn Sie einen Backfill mit einem externen Tool durchführen möchten, oder CDC über eine von Ihnen angegebene Logdatei starten. Bei einer MySQL-Quelle müssen Sie eine binlog-Position und bei einer Oracle-Quelle eine System Change Number (SCN) in der Redo-Logdatei angeben.

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

Starten Sie einen Stream ab einer bestimmten binlog-Position (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 aus Sie den Stream starten möchten.
  • POSITION: Die Position in der Logdatei, ab der Sie den Stream starten möchten. Wenn Sie den Wert nicht angeben, beginnt Datastream mit dem Lesen aus dem Header 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 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.

Starten 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 Change Number (SCN) in der Redo-Logdatei, von der aus 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 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 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 zur Verwendung von gcloud zum Pausieren deines Streams 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 zur Verwendung von gcloud zum Fortsetzen des Streams finden Sie hier.

Stream wiederherstellen

Mit der Methode RunStream können Sie einen dauerhaft ausgefallenen 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 zur Wiederherstellung eines Streams für eine MySQL- oder Oracle-Quelle aus verschiedenen Positionen der Logdateien:

REST

Stelle einen Stream von der aktuellen Position wieder her. Das ist die Standardoption:

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

So stellen Sie einen Stream von 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 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 keinen Wert 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
      }
    }
  }
}

So stellen Sie einen Stream von einer bestimmten Position wieder her (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 Change Number (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 ä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 Einbinden einer UnifiedTypes-Schemadatei in den Dateipfad, den 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 dazu, wie Sie Ihren Stream mithilfe von gcloud ändern, 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 für den ersten Backfill für ein Objekt Ihres Streams finden Sie hier.

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 eine OBJECT_ID, die das Objekt eindeutig identifiziert. Sie verwenden die OBJECT_ID, um den Backfill für den Stream zu 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, um den Backfill für ein Objekt Ihres Streams zu beenden, finden Sie hier.

Maximale Anzahl gleichzeitiger CDC-Tasks ändern

Der folgende Code zeigt, wie Sie die Anzahl der maximalen gleichzeitigen Change Data Capture-Aufgaben (CDC) 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 mit 0 definieren, wird der Systemstandard von 5 Aufgaben für den Stream festgelegt.

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 Sie die Anzahl der maximalen gleichzeitigen Backfill-Aufgaben für einen MySQL-Stream auf 25 festlegen können.

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

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 wie Binary Large Objects (BLOB), „Character Large Objects“ (CLOB) und „National Character Large Objects“ (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 erstellen, mit dem Sie große Objekte streamen können.

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.