Auf dieser Seite wird gezeigt, wie Sie Dataflow-Pipelines erstellen, die Spanner-Änderungsdaten mithilfe von Änderungsstreams nutzen und weiterleiten. Mit dem Beispielcode auf dieser Seite können Sie benutzerdefinierte Pipelines erstellen.
Wichtige Konzepte
Im Folgenden finden Sie einige Kernkonzepte für Dataflow-Pipelines für Änderungsstreams.
Dataflow
Dataflow ist ein serverloser, schneller und kostengünstiger Dienst, der sowohl Stream- als auch Batchverarbeitung unterstützt. Er bietet Portabilität mit Verarbeitungsjobs, die in den Open-Source-Apache Beam-Bibliotheken geschrieben wurden, und automatisiert die Infrastrukturbereitstellung und die Clusterverwaltung. Dataflow bietet ein Streaming nahezu in Echtzeit beim Lesen aus Änderungsstreams.
Sie können Dataflow verwenden, um Spanner-Änderungsstreams mit dem SpannerIO-Connector zu verarbeiten, der eine Abstraktion über die Spanner API zum Abfragen von Änderungsstreams bietet. Mit diesem Connector müssen Sie den Partitionslebenszyklus für Änderungsstreams nicht verwalten. Dies ist erforderlich, wenn Sie die Spanner API direkt verwenden. Der Connector stellt Ihnen einen Strom von Änderungsdatensätzen zur Verfügung, sodass Sie sich mehr auf die Anwendungslogik konzentrieren können und weniger auf bestimmte API-Details und die dynamische Partitionierung des Änderungsstreams. In den meisten Fällen, in denen Änderungsstreamdaten gelesen werden müssen, empfehlen wir die Verwendung des SpannerIO-Connectors anstelle der Spanner API.
Dataflow-Vorlagen sind vordefinierte Dataflow-Pipelines, die gängige Anwendungsfälle implementieren. Eine Übersicht finden Sie unter Dataflow-Vorlagen.
Dataflow-Pipeline
Eine Dataflow-Pipeline für Spanner-Änderungsstreams besteht aus vier Hauptteilen:
- Eine Spanner-Datenbank mit einem Änderungsstream
- Der SpannerIO-Connector
- Benutzerdefinierte Transformationen und Senken
- Ein Senken-E/A-Autor
Diese werden im Folgenden ausführlicher erörtert.
Spanner-Änderungsstream
Weitere Informationen zum Erstellen eines Änderungsstreams finden Sie unter Änderungsstream erstellen.
Apache Beam SpannerIO-Connector
Dies ist der zuvor beschriebene SpannerIO-Connector. Es ist ein E/A-Quell-Connector, der eine PCollection an Änderungseinträgen für Daten für spätere Phasen der Pipeline ausgibt. Die Ereigniszeit für jeden ausgegebenen Änderungseintrag von Daten ist der Commit-Zeitstempel. Beachten Sie, dass die ausgegebenen Datensätze ungeordnet sind und der SpannerIO-Connector garantiert, dass es keine verspäteten Datensätze gibt.
Bei der Arbeit mit Änderungsstreams verwendet Dataflow Prüfpunktausführung. Daher kann jeder Worker bis zum konfigurierten Prüfpunktintervall auf die Zwischenspeicherung von Änderungen warten, bevor er die Änderungen zur weiteren Verarbeitung sendet.
Benutzerdefinierte Transformationen
Mit einer benutzerdefinierten Transformation kann ein Nutzer Verarbeitungsdaten in einer Dataflow-Pipeline aggregieren, transformieren oder ändern. Gängige Anwendungsfälle hierfür sind das Entfernen personenidentifizierbarer Informationen, die Erfüllung der Formatanforderungen für nachgelagerte Daten und das Sortieren. In der offiziellen Apache Beam-Dokumentation finden Sie einen Programmierleitfaden zu transforms.
Apache Beam-Senken-E/A-Autor
Apache Beam enthält integrierte E/A-Transformationen, mit denen aus einer Dataflow-Pipeline in eine Datensenke wie BigQuery geschrieben werden kann. Die gängigsten Datensenken werden nativ unterstützt.
Dataflow-Vorlagen
Dataflow-Vorlagen bieten eine einfache Möglichkeit, Dataflow-Jobs basierend auf vordefinierten Docker-Images für gängige Anwendungsfälle über die Google Cloud Console, die Google Cloud CLI oder REST API-Aufrufe zu erstellen.
Für Spanner-Änderungsstreams stehen drei flexible Dataflow-Vorlagen zur Verfügung:
Dataflow-Pipeline erstellen
In diesem Abschnitt wird die Erstkonfiguration des Connectors beschrieben. Außerdem finden Sie hier Beispiele für gängige Integrationen in die Spanner-Funktion für Änderungsstreams.
Zum Ausführen dieser Schritte benötigen Sie eine Java-Entwicklungsumgebung für Dataflow. Weitere Informationen finden Sie unter Dataflow-Pipeline mit Java erstellen.
Änderungsstream erstellen
Weitere Informationen zum Erstellen eines Änderungsstreams finden Sie unter Änderungsstream erstellen. Sie benötigen eine Spanner-Datenbank mit einem konfigurierten Änderungsstream, um mit den nächsten Schritten fortfahren zu können.
Detaillierte Berechtigungen für die Zugriffssteuerung gewähren
Wenn Sie erwarten, dass Nutzer mit einer detaillierten Zugriffssteuerung den Dataflow-Job ausführen, müssen Sie ihnen Zugriff auf eine Datenbankrolle gewähren, die die Berechtigung SELECT für den Änderungsstream und die Berechtigung EXECUTE für die Tabellenwertfunktion des Änderungsstreams hat. Achten Sie außerdem darauf, dass das Hauptkonto die Datenbankrolle in der SpannerIO-Konfiguration oder in der flexiblen Dataflow-Vorlage angibt.
Weitere Informationen finden Sie unter Detaillierte Zugriffssteuerung.
SpannerIO-Connector als Abhängigkeit hinzufügen
Der Apache Beam SpannerIO-Connector kapselt die Komplexität der Verarbeitung der Änderungsstreams direkt über die Cloud Spanner API und gibt eine Kollektion von Änderungsstream-Datensätzen an spätere Phasen der Pipeline aus.
Diese Objekte können auch in anderen Phasen der Dataflow-Pipeline des Nutzers aufgenommen werden. Die Änderungsstream-Integration ist Teil des SpannerIO-Connectors. Um den SpannerIO-Connector verwenden zu können, muss die Abhängigkeit zu Ihrer pom.xml-Datei hinzugefügt werden:
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
<version>${beam-version}</version> <!-- available from version 2.38.0 -->
</dependency>
Metadatendatenbank erstellen
Der Connector muss beim Ausführen der Apache Beam-Pipeline jede Partition erfassen. Diese Metadaten werden in einer Spanner-Tabelle gespeichert, die vom Connector während der Initialisierung erstellt wird. Sie geben die Datenbank an, in der diese Tabelle erstellt wird, wenn Sie den Connector konfigurieren.
Wie unter Best Practices für Änderungsstreams beschrieben, empfehlen wir, zu diesem Zweck eine neue Datenbank zu erstellen, anstatt dem Connector zu erlauben, die Datenbank Ihrer Anwendung zum Speichern der Metadatentabelle zu verwenden.
Der Inhaber eines Dataflow-Jobs, der den SpannerIO-Connector verwendet, muss die folgenden IAM-Berechtigungen für diese Metadatendatenbank festgelegt haben:
spanner.databases.updateDdlspanner.databases.beginReadOnlyTransactionspanner.databases.beginOrRollbackReadWriteTransactionspanner.databases.readspanner.databases.selectspanner.databases.writespanner.sessions.createspanner.sessions.get
Connector konfigurieren
Der Connector für Spanner-Änderungsstreams kann so konfiguriert werden:
SpannerConfig spannerConfig = SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role"); // Needed for fine-grained access control only
Timestamp startTime = Timestamp.now();
Timestamp endTime = Timestamp.ofTimeSecondsAndNanos(
startTime.getSeconds() + (10 * 60),
startTime.getNanos()
);
SpannerIO
.readChangeStream()
.withSpannerConfig(spannerConfig)
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-meta-instance-id")
.withMetadataDatabase("my-meta-database-id")
.withMetadataTable("my-meta-table-name")
.withRpcPriority(RpcPriority.MEDIUM)
.withInclusiveStartAt(startTime)
.withInclusiveEndAt(endTime);
Im Folgenden werden die readChangeStream()-Optionen beschrieben:
Spanner-Konfiguration (erforderlich)
Wird zum Konfigurieren des Projekts, der Instanz und der Datenbank verwendet, in denen der Änderungsstream erstellt wurde und von wo aus abgefragt werden soll. Gibt optional auch die Datenbankrolle an, die verwendet werden soll, wenn das IAM-Hauptkonto, das den Dataflow-Job ausführt, ein Nutzer mit detaillierter Zugriffssteuerung ist. Der Job übernimmt diese Datenbankrolle für den Zugriff auf den Änderungsstream. Weitere Informationen finden Sie unter Detaillierte Zugriffssteuerung.
Name des Streams ändern (erforderlich)
Mit diesem Namen wird der Änderungsstream eindeutig identifiziert. Der hier angegebene Name muss mit dem Namen übereinstimmen, der bei der Erstellung verwendet wurde.
Metadateninstanz-ID (optional)
Dies ist die Instanz zum Speichern der Metadaten, die vom Connector verwendet werden, um die Nutzung der API-Daten des Änderungsstreams zu steuern.
ID der Metadatendatenbank (erforderlich)
Dies ist die Datenbank zum Speichern der Metadaten, die vom Connector verwendet werden, um die Nutzung der API-Daten des Änderungsstreams zu steuern.
Name der Metadatentabelle (optional)
Sollte nur beim Aktualisieren einer vorhandenen Pipeline verwendet werden.
Das ist der bereits vorhandene Name der Metadatentabelle, der vom Connector verwendet wird. Er wird vom Connector verwendet, um die Metadaten zu speichern und so die Nutzung der API-Daten des Änderungsstreams zu steuern. Wenn diese Option nicht angegeben wird, erstellt Spanner bei der Initialisierung des Connectors eine neue Tabelle mit einem generierten Namen.
RPC-Priorität (optional)
Die Anfragepriorität, die für die Abfragen des Änderungsstreams verwendet werden soll. Wenn Sie diesen Parameter nicht angeben, wird high
priority verwendet.
InclusiveStartAt (erforderlich)
Änderungen ab dem angegebenen Zeitstempel werden an den Aufrufer zurückgegeben.
InclusiveEndAt (optional)
Änderungen bis zum angegebenen Zeitstempel werden an den Aufrufer zurückgegeben. Wenn Sie diesen Parameter nicht angeben, werden Änderungen auf unbestimmte Zeit ausgegeben.
Transformationen und Senken hinzufügen, um Änderungsdaten zu verarbeiten
Nachdem die vorherigen Schritte abgeschlossen sind, kann der konfigurierte SpannerIO-Connector eine Kollektion von DataChangeRecord-Objekten ausgeben.
Unter Beispieltransformationen und -senken finden Sie mehrere Beispiel-Pipeline-Konfigurationen, die diese gestreamten Daten auf verschiedene Weise verarbeiten.
Beachten Sie, dass die vom SpannerIO-Connector ausgegebenen Änderungsstreameinträge ungeordnet sind. Das liegt daran, dass PCollections keine Bestellgarantien bieten. Wenn Sie einen geordneten Stream benötigen, müssen Sie die Datensätze in Ihren Pipelines als Transformationen gruppieren und sortieren. Siehe Beispiel: Nach Schlüssel sortieren. Sie können dieses Beispiel erweitern, um die Datensätze nach beliebigen Feldern der Datensätze zu sortieren, z. B. nach Transaktions-IDs.
Beispieltransformationen und -senken
Sie können eigene Transformationen definieren und Senken angeben, in die die Daten geschrieben werden. Die Apache Beam-Dokumentation bietet eine Vielzahl von transforms, die angewendet werden können, sowie gebrauchsfertige E/A-Connectors, um die Daten in externe Systeme zu schreiben.
Beispiel: Sortieren nach Schlüssel
In diesem Codebeispiel werden mithilfe des Dataflow-Connectors Datenänderungseinträge ausgegeben, die nach Commit-Zeitstempel sortiert und nach Primärschlüsseln gruppiert sind.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new BreakRecordByModFn()))
.apply(ParDo.of(new KeyByIdFn()))
.apply(ParDo.of(new BufferKeyUntilOutputTimestamp()))
// Subsequent processing goes here
Dieses Codebeispiel verwendet Status und Timer, um Datensätze für jeden Schlüssel zu puffern, und legt die Ablaufzeit des Timers auf eine vom Nutzer konfigurierte Zeit T in der Zukunft fest (definiert in der Funktion BufferKeyUntilOutputTimestamp). Wenn das Dataflow-Wasserzeichen die Zeit T durchläuft, leert dieser Code alle Datensätze im Zwischenspeicher mit einem Zeitstempel kleiner als T, ordnet diese Datensätze nach dem Commit-Zeitstempel und gibt ein Schlüssel/Wert-Paar aus, wobei Folgendes gilt:
- Der Schlüssel ist der Eingabeschlüssel, also der Primärschlüssel, der in einem Bucket-Array der Größe 1.000 gehasht ist.
- Der Wert sind die sortierten Datenänderungseinträge, die für den Schlüssel zwischengespeichert wurden.
Für jeden Schlüssel gibt es folgende Garantien:
- Timer werden garantiert in der Reihenfolge des Ablaufzeitstempels ausgelöst.
- Nachgelagerte Phasen erhalten die Elemente garantiert in derselben Reihenfolge, in der sie produziert wurden.
Beispiel: Bei einem Schlüssel mit dem Wert 100 wird der Timer bei T1 bzw. T10 ausgelöst und es wird ein Bundle von Datenänderungseinträgen bei jedem Zeitstempel erzeugt. Da die am T1 ausgegebenen Datenänderungseinträge vor den am T10 ausgegebenen Datenänderungseinträgen erstellt wurden, werden die am T1 ausgegebenen Datenänderungseinträge auch von der nächsten Phase vor den an T10 ausgegebenen Datenänderungseinträgen empfangen. Dieser Mechanismus hilft uns dabei, für die nachgelagerte Verarbeitung eine strikte Reihenfolge der Commit-Zeitstempel nach Primärschlüssel zu garantieren.
Dieser Vorgang wird so lange wiederholt, bis die Pipeline endet und alle Datensätze für Datenänderungen verarbeitet wurden. Wenn kein Ende angegeben ist, wird der Vorgang auf unbestimmte Zeit wiederholt.
Beachten Sie, dass in diesem Codebeispiel Status und Timer anstelle von Fenstern verwendet werden, um eine Sortierung nach Schlüssel durchzuführen. Die Begründung ist, dass Fenster nicht garantiert in der richtigen Reihenfolge verarbeitet werden. Dies bedeutet, dass ältere Fenster später verarbeitet werden können als neuere Fenster, was zu einer Bearbeitung außerhalb der Reihenfolge führen kann.
BreakRecordByModFn
Jeder Datensatz für eine Datenänderung kann mehrere Mods enthalten. Jeder Mod steht für das Einfügen, Aktualisieren oder Löschen eines einzelnen Primärschlüsselwerts. Diese Funktion teilt jeden Datensatz für Datenänderungen in separate Datensätze auf, einen pro Mod.
private static class BreakRecordByModFn extends DoFn<DataChangeRecord,
DataChangeRecord> {
@ProcessElement
public void processElement(
@Element DataChangeRecord record, OutputReceiver<DataChangeRecord>
outputReceiver) {
record.getMods().stream()
.map(
mod ->
new DataChangeRecord(
record.getPartitionToken(),
record.getCommitTimestamp(),
record.getServerTransactionId(),
record.isLastRecordInTransactionInPartition(),
record.getRecordSequence(),
record.getTableName(),
record.getRowType(),
Collections.singletonList(mod),
record.getModType(),
record.getValueCaptureType(),
record.getNumberOfRecordsInTransaction(),
record.getNumberOfPartitionsInTransaction(),
record.getTransactionTag(),
record.isSystemTransaction(),
record.getMetadata()))
.forEach(outputReceiver::output);
}
}
KeyByIdFn
Diese Funktion nimmt eine DataChangeRecord an und gibt eine DataChangeRecord aus, die durch den Spanner-Primärschlüssel verschlüsselt ist und als Ganzzahlwert gehasht ist.
private static class KeyByIdFn extends DoFn<DataChangeRecord, KV<String, DataChangeRecord>> {
// NUMBER_OF_BUCKETS should be configured by the user to match their key cardinality
// Here, we are choosing to hash the Spanner primary keys to a bucket index, in order to have a deterministic number
// of states and timers for performance purposes.
// Note that having too many buckets might have undesirable effects if it results in a low number of records per bucket
// On the other hand, having too few buckets might also be problematic, since many keys will be contained within them.
private static final int NUMBER_OF_BUCKETS = 1000;
@ProcessElement
public void processElement(
@Element DataChangeRecord record,
OutputReceiver<KV<String, DataChangeRecord>> outputReceiver) {
int hashCode = (int) record.getMods().get(0).getKeysJson().hashCode();
// Hash the received keys into a bucket in order to have a
// deterministic number of buffers and timers.
String bucketIndex = String.valueOf(hashCode % NUMBER_OF_BUCKETS);
outputReceiver.output(KV.of(bucketIndex, record));
}
}
BufferKeyUntilOutputTimestamp
Timer und Puffer sind pro Schlüssel. Diese Funktion puffert jeden Datenänderungseintrag, bis das Wasserzeichen den Zeitstempel passiert, bei dem die zwischengespeicherten Datenänderungseinträge ausgegeben werden sollen.
Dieser Code verwendet einen Timer in einer Schleife, um zu bestimmen, wann der Zwischenspeicher geleert werden soll:
- Wird zum ersten Mal ein Datenänderungseintrag für einen Schlüssel erkannt, wird der Timer so eingestellt, dass er beim Commit-Zeitstempel des Datenänderungseintrags +
incrementIntervalSeconds(eine vom Nutzer konfigurierbare Option) ausgelöst wird. - Wenn der Timer ausgelöst wird, werden im Zwischenspeicher alle Datensätze für Datenänderungen hinzugefügt, deren Zeitstempel kleiner als die Ablaufzeit des Timers ist (
recordsToOutput). Wenn der Puffer über Datenänderungseinträge verfügt, deren Zeitstempel größer oder gleich der Ablaufzeit des Timers ist, fügt er diese Datenänderungseinträge wieder in den Puffer hinzu, anstatt sie auszugeben. Anschließend wird der nächste Timer auf die Ablaufzeit des aktuellen Timers plusincrementIntervalInSecondsgesetzt. - Wenn
recordsToOutputnicht leer ist, ordnet die Funktion die Datensätze für Datenänderungen inrecordsToOutputnach dem Commit-Zeitstempel und der Transaktions-ID und gibt sie dann aus.
private static class BufferKeyUntilOutputTimestamp extends
DoFn<KV<String, DataChangeRecord>, KV<String, Iterable<DataChangeRecord>>> {
private static final Logger LOG =
LoggerFactory.getLogger(BufferKeyUntilOutputTimestamp.class);
private final long incrementIntervalInSeconds = 2;
private BufferKeyUntilOutputTimestamp(long incrementIntervalInSeconds) {
this.incrementIntervalInSeconds = incrementIntervalInSeconds;
}
@SuppressWarnings("unused")
@TimerId("timer")
private final TimerSpec timerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
@StateId("buffer")
private final StateSpec<BagState<DataChangeRecord>> buffer = StateSpecs.bag();
@StateId("keyString")
private final StateSpec<ValueState<String>> keyString =
StateSpecs.value(StringUtf8Coder.of());
@ProcessElement
public void process(
@Element KV<String, DataChangeRecord> element,
@StateId("buffer") BagState<DataChangeRecord> buffer,
@TimerId("timer") Timer timer,
@StateId("keyString") ValueState<String> keyString) {
buffer.add(element.getValue());
// Only set the timer if this is the first time we are receiving a data change
// record with this key.
String elementKey = keyString.read();
if (elementKey == null) {
Instant commitTimestamp =
new Instant(element.getValue().getCommitTimestamp().toSqlTimestamp());
Instant outputTimestamp =
commitTimestamp.plus(Duration.standardSeconds(incrementIntervalInSeconds));
timer.set(outputTimestamp);
keyString.write(element.getKey());
}
}
@OnTimer("timer")
public void onExpiry(
OnTimerContext context,
@StateId("buffer") BagState<DataChangeRecord> buffer,
@TimerId("timer") Timer timer,
@StateId("keyString") ValueState<String> keyString) {
if (!buffer.isEmpty().read()) {
String elementKey = keyString.read();
final List<DataChangeRecord> records =
StreamSupport.stream(buffer.read().spliterator(), false)
.collect(Collectors.toList());
buffer.clear();
List<DataChangeRecord> recordsToOutput = new ArrayList<>();
for (DataChangeRecord record : records) {
Instant recordCommitTimestamp =
new Instant(record.getCommitTimestamp().toSqlTimestamp());
final String recordString =
record.getMods().get(0).getNewValuesJson().isEmpty()
? "Deleted record"
: record.getMods().get(0).getNewValuesJson();
// When the watermark passes time T, this means that all records with
// event time < T have been processed and successfully committed. Since the
// timer fires when the watermark passes the expiration time, we should
// only output records with event time < expiration time.
if (recordCommitTimestamp.isBefore(context.timestamp())) {
LOG.info(
"Outputting record with key {} and value {} at expiration " +
"timestamp {}",
elementKey,
recordString,
context.timestamp().toString());
recordsToOutput.add(record);
} else {
LOG.info(
"Expired at {} but adding record with key {} and value {} back to " +
"buffer due to commit timestamp {}",
context.timestamp().toString(),
elementKey,
recordString,
recordCommitTimestamp.toString());
buffer.add(record);
}
}
// Output records, if there are any to output.
if (!recordsToOutput.isEmpty()) {
// Order the records in place, and output them. The user would need
// to implement DataChangeRecordComparator class that sorts the
// data change records by commit timestamp and transaction ID.
Collections.sort(recordsToOutput, new DataChangeRecordComparator());
context.outputWithTimestamp(
KV.of(elementKey, recordsToOutput), context.timestamp());
LOG.info(
"Expired at {}, outputting records for key {}",
context.timestamp().toString(),
elementKey);
} else {
LOG.info("Expired at {} with no records", context.timestamp().toString());
}
}
Instant nextTimer = context.timestamp().plus(Duration.standardSeconds(incrementIntervalInSeconds));
if (buffer.isEmpty() != null && !buffer.isEmpty().read()) {
LOG.info("Setting next timer to {}", nextTimer.toString());
timer.set(nextTimer);
} else {
LOG.info(
"Timer not being set since the buffer is empty: ");
keyString.clear();
}
}
}
Bestelltransaktionen
Diese Pipeline kann nach Transaktions-ID und Commit-Zeitstempel sortiert werden. Zwischenspeichern Sie dazu Einträge für jedes Transaktions-ID-/Commit-Zeitstempelpaar und nicht für jeden Spanner-Schlüssel. Hierfür muss der Code in KeyByIdFn geändert werden.
Beispiel: Transaktionen zusammenstellen
Dieses Codebeispiel liest Datenänderungseinträge, fasst alle Datenänderungseinträge derselben Transaktion in einem einzigen Element zusammen und gibt dieses Element aus. Beachten Sie, dass die von diesem Beispielcode ausgegebenen Transaktionen nicht nach dem Commit-Zeitstempel sortiert sind.
In diesem Codebeispiel werden Puffer verwendet, um Transaktionen aus Datenänderungseinträgen zusammenzustellen. Nach Erhalt eines Datenänderungseintrags, der zum ersten Mal zu einer Transaktion gehört, liest er das Feld numberOfRecordsInTransaction im Datenänderungseintrag, das die erwartete Anzahl von Datenänderungseinträgen beschreibt, die zu dieser Transaktion gehören. Sie puffert die zu dieser Transaktion gehörenden Datenänderungseinträge, bis die Anzahl der zwischengespeicherten Datensätze mit numberOfRecordsInTransaction übereinstimmt. Anschließend werden die gebündelten Datensätze zu Datenänderungen ausgegeben.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new KeyByTransactionIdFn()))
.apply(ParDo.of(new TransactionBoundaryFn()))
// Subsequent processing goes here
KeyByTransactionIdFn
Diese Funktion nimmt eine DataChangeRecord an und gibt eine DataChangeRecord aus, die durch die Transaktions-ID verschlüsselt ist.
private static class KeyByTransactionIdFn extends DoFn<DataChangeRecord, KV<String, DataChangeRecord>> {
@ProcessElement
public void processElement(
@Element DataChangeRecord record,
OutputReceiver<KV<String, DataChangeRecord>> outputReceiver) {
outputReceiver.output(KV.of(record.getServerTransactionId(), record));
}
}
TransactionBoundaryFn
TransactionBoundaryFn-Zwischenspeicher haben Schlüssel/Wert-Paare mit {TransactionId, DataChangeRecord} von KeyByTransactionIdFn empfangen und in Gruppen basierend auf TransactionId zwischengespeichert. Wenn die Anzahl der zwischengespeicherten Datensätze der Anzahl der Datensätze in der gesamten Transaktion entspricht, sortiert diese Funktion die DataChangeRecord-Objekte in der Gruppe nach Datensatzsequenz und gibt ein Schlüssel/Wert-Paar aus {CommitTimestamp, TransactionId}, Iterable<DataChangeRecord> aus.
Hier wird davon ausgegangen, dass SortKey eine benutzerdefinierte Klasse ist, die ein {CommitTimestamp, TransactionId}-Paar darstellt. Siehe Implementierungsbeispiel für SortKey.
private static class TransactionBoundaryFn extends DoFn<KV<String, DataChangeRecord>, KV<SortKey, Iterable<DataChangeRecord>>> {
@StateId("buffer")
private final StateSpec<BagState<DataChangeRecord>> buffer = StateSpecs.bag();
@StateId("count")
private final StateSpec<ValueState<Integer>> countState = StateSpecs.value();
@ProcessElement
public void process(
ProcessContext context,
@StateId("buffer") BagState<DataChangeRecord> buffer,
@StateId("count") ValueState<Integer> countState) {
final KV<String, DataChangeRecord> element = context.element();
final DataChangeRecord record = element.getValue();
buffer.add(record);
int count = (countState.read() != null ? countState.read() : 0);
count = count + 1;
countState.write(count);
if (count == record.getNumberOfRecordsInTransaction()) {
final List<DataChangeRecord> sortedRecords =
StreamSupport.stream(buffer.read().spliterator(), false)
.sorted(Comparator.comparing(DataChangeRecord::getRecordSequence))
.collect(Collectors.toList());
final Instant commitInstant =
new Instant(sortedRecords.get(0).getCommitTimestamp().toSqlTimestamp()
.getTime());
context.outputWithTimestamp(
KV.of(
new SortKey(sortedRecords.get(0).getCommitTimestamp(),
sortedRecords.get(0).getServerTransactionId()),
sortedRecords),
commitInstant);
buffer.clear();
countState.clear();
}
}
}
Beispiel: Nach Transaktions-Tag filtern
Wenn eine Transaktion, die Nutzerdaten ändert, getaggt wird, werden das entsprechende Tag und sein Typ als Teil von DataChangeRecord gespeichert. Die folgenden Beispiele veranschaulichen, wie Änderungsstream-Datensätze anhand von benutzerdefinierten Transaktions-Tags und System-Tags gefiltert werden:
Benutzerdefinierte Tag-Filterung für my-tx-tag:
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(Filter.by(record ->
!record.isSystemTransaction()
&& record.getTransactionTag().equalsIgnoreCase("my-tx-tag")))
// Subsequent processing goes here
System-Tag-Filterung/TTL-Prüfung:
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(Filter.by(record ->
record.isSystemTransaction()
&& record.getTransactionTag().equals("RowDeletionPolicy")))
// Subsequent processing goes here
Beispiel: Vollständige Zeile abrufen
In diesem Beispiel wird eine Spanner-Tabelle mit dem Namen Singer mit der folgenden Definition verwendet:
CREATE TABLE Singers (
SingerId INT64 NOT NULL,
FirstName STRING(1024),
LastName STRING(1024)
) PRIMARY KEY (SingerId);
Im standardmäßigen OLD_AND_NEW_VALUES-Werterfassungsmodus von Änderungsstreams enthält der empfangene Datenänderungseintrag bei einer Aktualisierung einer Spanner-Zeile nur die geänderten Spalten. Nachverfolgte, aber unveränderte Spalten werden nicht in den Datensatz aufgenommen. Mit dem Primärschlüssel des Mods kann am Commit-Zeitstempel des Datenänderungseintrags ein Spanner-Snapshot gelesen werden, um die unveränderten Spalten oder sogar die vollständige Zeile abzurufen.
Beachten Sie, dass die Datenbank-Aufbewahrungsrichtlinie möglicherweise in einen Wert geändert werden muss, der größer oder gleich der Aufbewahrungsrichtlinie des Änderungsstreams ist, damit der Snapshot-Lesevorgang erfolgreich ist.
Beachten Sie auch, dass die Verwendung des Werterfassungstyps NEW_ROW dafür die empfohlene und effizientere Methode ist, da standardmäßig alle verfolgten Spalten der Zeile zurückgegeben werden und kein zusätzlicher Snapshot-Lesevorgang in Spanner erforderlich ist.
SpannerConfig spannerConfig = SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role"); // Needed for fine-grained access control only
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(spannerConfig)
// Assume we have a change stream "my-change-stream" that watches Singers table.
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new ToFullRowJsonFn(spannerConfig)))
// Subsequent processing goes here
ToFullRowJsonFn
Diese Transformation führt am Commit-Zeitstempel jedes empfangenen Eintrags einen veralteten Lesevorgang durch und ordnet die vollständige Zeile JSON zu.
public class ToFullRowJsonFn extends DoFn<DataChangeRecord, String> {
// Since each instance of this DoFn will create its own session pool and will
// perform calls to Spanner sequentially, we keep the number of sessions in
// the pool small. This way, we avoid wasting resources.
private static final int MIN_SESSIONS = 1;
private static final int MAX_SESSIONS = 5;
private final String projectId;
private final String instanceId;
private final String databaseId;
private transient DatabaseClient client;
private transient Spanner spanner;
public ToFullRowJsonFn(SpannerConfig spannerConfig) {
this.projectId = spannerConfig.getProjectId().get();
this.instanceId = spannerConfig.getInstanceId().get();
this.databaseId = spannerConfig.getDatabaseId().get();
}
@Setup
public void setup() {
SessionPoolOptions sessionPoolOptions = SessionPoolOptions
.newBuilder()
.setMinSessions(MIN_SESSIONS)
.setMaxSessions(MAX_SESSIONS)
.build();
SpannerOptions options = SpannerOptions
.newBuilder()
.setProjectId(projectId)
.setSessionPoolOption(sessionPoolOptions)
.build();
DatabaseId id = DatabaseId.of(projectId, instanceId, databaseId);
spanner = options.getService();
client = spanner.getDatabaseClient(id);
}
@Teardown
public void teardown() {
spanner.close();
}
@ProcessElement
public void process(
@Element DataChangeRecord element,
OutputReceiver<String> output) {
com.google.cloud.Timestamp commitTimestamp = element.getCommitTimestamp();
element.getMods().forEach(mod -> {
JSONObject keysJson = new JSONObject(mod.getKeysJson());
JSONObject newValuesJson = new JSONObject(mod.getNewValuesJson());
ModType modType = element.getModType();
JSONObject jsonRow = new JSONObject();
long singerId = keysJson.getLong("SingerId");
jsonRow.put("SingerId", singerId);
if (modType == ModType.INSERT) {
// For INSERT mod, get non-primary key columns from mod.
jsonRow.put("FirstName", newValuesJson.get("FirstName"));
jsonRow.put("LastName", newValuesJson.get("LastName"));
} else if (modType == ModType.UPDATE) {
// For UPDATE mod, get non-primary key columns by doing a snapshot read using the primary key column from mod.
try (ResultSet resultSet = client
.singleUse(TimestampBound.ofReadTimestamp(commitTimestamp))
.read(
"Singers",
KeySet.singleKey(com.google.cloud.spanner.Key.of(singerId)),
Arrays.asList("FirstName", "LastName"))) {
if (resultSet.next()) {
jsonRow.put("FirstName", resultSet.isNull("FirstName") ?
JSONObject.NULL : resultSet.getString("FirstName"));
jsonRow.put("LastName", resultSet.isNull("LastName") ?
JSONObject.NULL : resultSet.getString("LastName"));
}
}
} else {
// For DELETE mod, there is nothing to do, as we already set SingerId.
}
output.output(jsonRow.toString());
});
}
}
Mit diesem Code wird ein Spanner-Datenbankclient erstellt, um den gesamten Zeilenabruf durchzuführen. Außerdem wird der Sitzungspool so konfiguriert, dass er nur wenige Sitzungen hat und Lesevorgänge in einer Instanz des ToFullRowJsonFn sequenziell ausführen. Dataflow erzeugt viele Instanzen dieser Funktion mit jeweils einem eigenen Clientpool.
Beispiel: Spanner zu Pub/Sub
In diesem Szenario streamt der Aufrufer die Datensätze so schnell wie möglich an Pub/Sub, ohne Gruppierung oder Aggregation. Dies eignet sich gut zum Auslösen der nachgelagerten Verarbeitung, da alle neuen Zeilen, die in eine Spanner-Tabelle eingefügt werden, zur weiteren Verarbeitung an Pub/Sub gestreamt werden.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(MapElements.into(TypeDescriptors.strings()).via(Object::toString))
.apply(PubsubIO.writeStrings().to("my-topic"));
Die Pub/Sub-Senke kann so konfiguriert werden, dass die Semantik genau einmal sichergestellt ist.
Beispiel: Spanner für Cloud Storage
In diesem Szenario gruppiert der Aufrufer alle Datensätze in einem bestimmten Fenster und speichert die Gruppe in separaten Cloud Storage-Dateien. Dies eignet sich gut für Analysen und die Archivierung zu einem bestimmten Zeitpunkt, die unabhängig von der Aufbewahrungsdauer von Spanner ist.
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role")) // Needed for fine-grained access control only
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(MapElements.into(TypeDescriptors.strings()).via(Object::toString))
.apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))))
.apply(TextIO
.write()
.to("gs://my-bucket/change-stream-results-")
.withSuffix(".txt")
.withWindowedWrites()
.withNumShards(1));
Die Cloud Storage-Senke bietet standardmäßig mindestens einmal Semantik. Mit zusätzlichem Verarbeitungsaufwand kann er so geändert werden, dass er eine genau einmalige Semantik hat.
Wir stellen auch eine Dataflow-Vorlage für diesen Anwendungsfall zur Verfügung: siehe Änderungsstreams mit Cloud Storage verbinden.
Beispiel: Spanner für BigQuery (Haupttabelle)
Hier streamt der Aufrufer Änderungseinträge in BigQuery. Jeder Datensatz zur Datenänderung wird in BigQuery als eine Zeile dargestellt. Dies ist eine gute Lösung für Analysen. Dieser Code verwendet die zuvor im Abschnitt Vollständige Zeile abrufen definierten Funktionen, um die vollständige Zeile des Eintrags abzurufen und in BigQuery zu schreiben.
SpannerConfig spannerConfig = SpannerConfig
.create()
.withProjectId("my-project-id")
.withInstanceId("my-instance-id")
.withDatabaseId("my-database-id")
.withDatabaseRole("my-database-role"); // Needed for fine-grained access control only
pipeline
.apply(SpannerIO
.readChangeStream()
.withSpannerConfig(spannerConfig)
.withChangeStreamName("my-change-stream")
.withMetadataInstance("my-metadata-instance-id")
.withMetadataDatabase("my-metadata-database-id")
.withInclusiveStartAt(Timestamp.now()))
.apply(ParDo.of(new ToFullRowJsonFn(spannerConfig)))
.apply(BigQueryIO
.<String>write()
.to("my-bigquery-table")
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(Write.WriteDisposition.WRITE_APPEND)
.withSchema(new TableSchema().setFields(Arrays.asList(
new TableFieldSchema()
.setName("SingerId")
.setType("INT64")
.setMode("REQUIRED"),
new TableFieldSchema()
.setName("FirstName")
.setType("STRING")
.setMode("REQUIRED"),
new TableFieldSchema()
.setName("LastName")
.setType("STRING")
.setMode("REQUIRED")
)))
.withAutoSharding()
.optimizedWrites()
.withFormatFunction((String element) -> {
ObjectMapper objectMapper = new ObjectMapper();
JsonNode jsonNode = null;
try {
jsonNode = objectMapper.readTree(element);
} catch (IOException e) {
e.printStackTrace();
}
return new TableRow()
.set("SingerId", jsonNode.get("SingerId").asInt())
.set("FirstName", jsonNode.get("FirstName").asText())
.set("LastName", jsonNode.get("LastName").asText());
}
)
);
Die BigQuery-Senke bietet standardmäßig mindestens einmal Semantik. Mit zusätzlichem Verarbeitungsaufwand kann er so geändert werden, dass er eine genau einmalige Semantik hat.
Wir stellen auch eine Dataflow-Vorlage für diesen Anwendungsfall zur Verfügung. Siehe Änderungsstreams mit BigQuery verbinden.
Pipeline überwachen
Zum Überwachen einer Dataflow-Pipeline für Änderungsstreams stehen zwei Klassen von Messwerten zur Verfügung.
Dataflow-Standardmesswerte
Dataflow bietet verschiedene Messwerte, um dafür zu sorgen, dass Ihr Job fehlerfrei ist, z. B. Datenaktualität, Systemverzögerung, Jobdurchsatz und Worker-CPU-Auslastung. Weitere Informationen finden Sie unter Monitoring für Dataflow-Pipelines verwenden.
Bei Pipelines für Änderungsstreams sind zwei Hauptmesswerte zu berücksichtigen: die Systemlatenz und die Datenaktualität.
Die Systemlatenz gibt Ihnen die aktuelle maximale Dauer (in Sekunden) an, für die ein Datenelement verarbeitet wird oder auf die Verarbeitung wartet.
Die Datenaktualität zeigt Ihnen die Zeitspanne zwischen jetzt (Echtzeit) und dem Ausgabewasserzeichen. Das Ausgabe-Wasserzeichen der Zeit T gibt an, dass alle Elemente mit einer Ereigniszeit (streng) vor T für die Berechnung verarbeitet wurden. Mit anderen Worten: Mit der Datenaktualität wird gemessen, wie aktuell die Pipeline im Hinblick auf die Verarbeitung der empfangenen Ereignisse ist.
Wenn für die Pipeline nicht genügend Ressourcen vorhanden sind, können Sie dies an diesen beiden Messwerten ablesen. Die Systemlatenz nimmt zu, da Elemente länger warten müssen, bevor sie verarbeitet werden können. Die Datenaktualität erhöht sich ebenfalls, da die Pipeline nicht mit der empfangenen Datenmenge Schritt halten kann.
Benutzerdefinierte Messwerte für Änderungsstreams
Diese Messwerte werden in Cloud Monitoring bereitgestellt und umfassen:
- Bucket-Latenz (Histogramm) zwischen einem Datensatz, der in Spanner per Commit übergeben wird, bis zu dem vom Connector an eine PCollection ausgegeben wird. Dieser Messwert kann verwendet werden, um alle Leistungsprobleme (Latenz) mit der Pipeline anzusehen.
- Gesamtzahl der gelesenen Datensätze. Dieser Wert gibt an, wie viele Datensätze der Connector insgesamt ausgegeben hat. Diese Zahl sollte kontinuierlich steigen und den Trend der Schreibvorgänge in der zugrunde liegenden Spanner-Datenbank widerspiegeln.
- Anzahl der Partitionen, die gerade gelesen werden. Es sollten immer Partitionen sein, die gelesen werden. Wenn diese Zahl null ist, ist ein Fehler in der Pipeline aufgetreten.
- Gesamtzahl der Abfragen, die während der Ausführung des Connectors ausgeführt werden. Dies ist ein allgemeiner Hinweis auf Änderungsstreamabfragen, die während der Ausführung der Pipeline an die Spanner-Instanz gestellt werden. Damit lässt sich eine Schätzung der Last vom Connector auf die Spanner-Datenbank abrufen.
Vorhandene Pipeline aktualisieren
Es ist möglich, eine laufende Pipeline, die den SpannerIO-Connector zum Verarbeiten von Änderungsstreams verwendet, zu aktualisieren, wenn die Jobkompatibilitätsprüfungen bestanden wurden. Dazu müssen Sie den Parameter für den Namen der Metadatentabelle des neuen Jobs beim Aktualisieren explizit festlegen. Verwenden Sie den Wert der Pipelineoption metadataTable aus dem Job, den Sie aktualisieren.
Wenn Sie eine von Google bereitgestellte Dataflow-Vorlage verwenden, legen Sie den Tabellennamen mit dem Parameter spannerMetadataTableName fest. Sie können den vorhandenen Job auch so ändern, dass die Metadatentabelle explizit mit der Methode withMetadataTable(your-metadata-table-name) in der Connector-Konfiguration verwendet wird. Anschließend können Sie der Anleitung unter Ersatzjob starten in der Dataflow-Dokumentation folgen, um einen laufenden Job zu aktualisieren.
Best Practices für Änderungsstreams und Dataflow
Im Folgenden finden Sie einige Best Practices zum Erstellen von Verbindungen zu Änderungsstreams mithilfe von Dataflow.
Separate Metadatendatenbank verwenden
Wir empfehlen, eine separate Datenbank für den SpannerIO-Connector zum Speichern von Metadaten zu erstellen, anstatt sie für die Verwendung Ihrer Anwendungsdatenbank zu konfigurieren.
Weitere Informationen finden Sie unter Separate Metadatendatenbank verwenden.
Größe des Clusters anpassen
Als Faustregel für die anfängliche Anzahl von Workern in einem Spanner-Änderungsstreamjob gilt ein Worker pro 1.000 Schreibvorgänge pro Sekunde. Diese Schätzung kann in Abhängigkeit von verschiedenen Faktoren variieren, z. B. von der Größe jeder Transaktion, der Anzahl der Änderungsstream-Datensätze, die aus einer einzelnen Transaktion und anderen Transformationen, Aggregationen oder Senken erstellt werden, die in der Pipeline verwendet werden.
Nach der ersten Ressourcenbeschaffung sollten Sie die unter Pipeline überwachen genannten Messwerte im Auge behalten, um sicherzustellen, dass die Pipeline fehlerfrei ist. Wir empfehlen, mit einer anfänglichen Worker-Pool-Größe zu experimentieren und zu überwachen, wie Ihre Pipeline mit der Last umgeht. Erhöhen Sie bei Bedarf die Anzahl der Knoten. Die CPU-Auslastung ist ein wichtiger Messwert, um zu prüfen, ob die Auslastung korrekt ist und mehr Knoten benötigt werden.
Bekannte Einschränkungen
Autoscaling
Die Autoscaling-Unterstützung für Pipelines, die SpannerIO.readChangeStream enthalten, erfordert Apache Beam 2.39.0 oder höher.
Wenn Sie eine Apache Beam-Version vor 2.39.0 verwenden, müssen Pipelines, die SpannerIO.readChangeStream enthalten, den Autoscaling-Algorithmus explizit als NONE angeben, wie unter Horizontales Autoscaling beschrieben.
Informationen zum manuellen Skalieren einer Dataflow-Pipeline anstelle des Autoscalings finden Sie unter Streamingpipeline manuell skalieren.
Runner V2
Für den Connector für Spanner-Änderungsstreams ist Dataflow Runner V2 erforderlich.
Diese muss während der Ausführung manuell angegeben werden, sonst wird ein Fehler ausgegeben. Sie können Runner V2 angeben, indem Sie Ihren Job mit --experiments=use_unified_worker,use_runner_v2 konfigurieren.
Snapshot
Der Connector für Spanner-Änderungsstreams unterstützt keine Dataflow-Snapshots.
Ausgleichen
Der Connector für Spanner-Änderungsstreams unterstützt das Leeren von Jobs nicht. Nur vorhandene Jobs können abgebrochen werden.
Sie können auch eine vorhandene Pipeline aktualisieren, ohne sie beenden zu müssen.
OpenCensus
Wenn Sie Ihre Pipeline mit OpenCensus überwachen möchten, geben Sie mindestens Version 0.28.3 an.
NullPointerException beim Pipelinestart
Ein Programmfehler in der Apache Beam-Version 2.38.0 kann unter bestimmten Bedingungen zu einem NullPointerException führen, wenn die Pipeline gestartet wird. Dies würde verhindern, dass Ihr Job gestartet wird, und stattdessen diese Fehlermeldung angezeigt:
java.lang.NullPointerException: null value in entry: Cloud Storage_PROJECT_ID=null
Verwenden Sie zum Beheben dieses Problems entweder die Apache Beam-Version 2.39.0 oder höher oder geben Sie die Version von beam-sdks-java-core manuell als 2.37.0 an:
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>2.37.0</version>
</dependency>