Daten mit Datastream und Dataflow in BigQuery streamen

Wenn Sie Daten mit der Streamingpipeline von Datastream und Dataflow in BigQuery replizieren, können Sie anpassen, wie Dataflow Änderungen verarbeitet und zusammenführt. Durch das Partitionieren von Replikattabellen mit benutzerdefinierten Schlüsseln, das Transformieren von Ereignissen mit benutzerdefinierten JavaScript-Funktionen (User-Defined Functions, UDFs) und das Anpassen der Zusammenführungsfrequenz können Sie ein Gleichgewicht zwischen Datenaktualität und Abfrage- und Speicherkosten finden.

Replikadatasets mit benutzerdefinierten Schlüsseln partitionieren

Das Staging-Dataset in BigQuery wird automatisch partitioniert. Standardmäßig wird das Replikadataset jedoch nicht partitioniert, da die Partitionierungsschlüssel für die Replikattabellen basierend auf einer bestimmten Geschäftslogik definiert werden müssen und nicht von Datastream und Dataflow erzwungen werden.

So partitionieren Sie jede Tabelle im Replikadataset:

1. Dataflow-Job beenden und leeren

2. Verwenden Sie den SQL-Editor in BigQuery, um das folgende SQL-Skript für jede Tabelle im Replikadataset auszuführen. In diesem Beispiel hat die Tabelle actor im Dataset datastream_cdc eine Spalte last_update, die wir als Partitionierungsschlüssel festlegen möchten. Durch Ausführen des Skripts erstellen Sie die Tabelle mit dem richtigen Partitionierungsschlüssel neu.

   create table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' partition by date(last_update)
   as SELECT * FROM '[BigQuery_PROJECT_ID].datastream_cdc.actor'
   
   drop table '[BigQuery_PROJECT_ID].datastream_cdc.actor'
   
   alter table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' rename to 'actor'

3. Verwenden Sie die Vorlage „Datastream für BigQuery “, um einen Dataflow-Job neu zu erstellen.

Benutzerdefinierte Funktionen zum Bearbeiten von Ereignisdaten ausführen

Sie können die Vorlage „Datastream für BigQuery“ verwenden, um eine benutzerdefinierte JavaScript-Funktion auszuführen. Dazu müssen Sie zuerst eine Datei mit der Funktion an einem bestimmten Speicherort in Cloud Storage platzieren. Gehen Sie anschließend so vor:

• Verwenden Sie den Parameter javascriptTextTransformGcsPath in der Vorlage, um den Speicherort der Datei in Cloud Storage anzugeben, die Ihre benutzerdefinierte Funktion enthält.
• Verwenden Sie den Parameter javascriptTextTransformFunctionName, um den Namen der JavaScript-Funktion anzugeben, die Sie als benutzerdefinierte Funktion aufrufen möchten.

Sie können beispielsweise eine benutzerdefinierte Funktion ausführen, um gelöschte Datensätze in den Tabellen des Replikadatasets in BigQuery beizubehalten. Dieser Vorgang wird als Soft-Delete bezeichnet.

Erstellen Sie dazu eine Funktion, die den Wert der Spalte _metadata_deleted in eine neue Spalte mit dem Namen is_deleted kopiert und dann den Wert der Spalte _metadata_deleted auf false zurücksetzt. Dadurch werden die Löschereignisse vom Dataflow-Job ignoriert und die gelöschten Datensätze beim Aktualisieren des Replikadatasets in BigQuery beibehalten.

Hier ist der Beispielcode für diese benutzerdefinierte Funktion:

/**
* This function changes the behavior of the Datastream to
* BigQuery template to allow soft deletes.
* @param {string} messageString from DatastreamIO data
* @return {string} same as an input message with an added property
*/
function transform(messageString) {
   // messageString is a JSON object as a string
   var messageJson = JSON.parse(messageString);
    // Moving the deleted flag to a new column will cause the pipeline to soft delete data.
   messageJson['is_deleted'] = messageJson['_metadata_deleted'];
   messageJson['_metadata_deleted'] = false;
    return JSON.stringify(messageJson);
 }

Zusammenführungsfrequenz festlegen

Verwenden Sie den Parameter mergeFrequencyMinutes der Vorlage „Datastream für BigQuery“, um die Zusammenführungsfrequenz festzulegen. Dies ist die Anzahl der Minuten zwischen den Zusammenführungen für eine bestimmte Tabelle im Replikadataset in BigQuery. Während Verlaufsdaten nachgefüllt werden, empfehlen wir, die Zusammenführungsfrequenz niedrig zu halten (12 oder 24 Stunden), um die Kosten unter Kontrolle zu halten.

Wenn Sie den Wert dieses Parameters beispielsweise auf 10 Minuten festlegen, führt Dataflow den Job, der die Vorlage verwendet, alle 10 Minuten aus. Bei der ersten Ausführung des Jobs gibt es jedoch eine Verzögerung von 5 Minuten. Wenn der Job beispielsweise um 9:14 Uhr ausgeführt wird, erfolgt die erste Zusammenführung um 9:29 Uhr (10 Minuten für die Zusammenführung und 5 Minuten für die Verzögerung). Die zweite Zusammenführung erfolgt um 9:39 Uhr und alle nachfolgenden Zusammenführungen erfolgen in Abständen von 10 Minuten (9:49 Uhr, 9:59 Uhr, 10:09 Uhr usw.).

Wenn Sie die Zusammenführungsfrequenz auf 60 Minuten festlegen, wird der Job nach einer Verzögerung von 5 Minuten zur vollen Stunde ausgeführt. Wenn der Job für 10:00 Uhr geplant ist, wird er aufgrund der Verzögerung von 5 Minuten tatsächlich um 10:05 Uhr ausgeführt. Alle nachfolgenden Zusammenführungen erfolgen in Abständen von 60 Minuten (11:05 Uhr, 12:05 Uhr, 13:05 Uhr usw.).

Aus Kostengründen oder anderen Gründen können Sie möglicherweise keine Zusammenführung mit einer Frequenz durchführen, die Ihren geschäftlichen Anforderungen entspricht. Möglicherweise haben Sie nicht die aktuellsten Daten. Um auf die aktuellsten Daten zuzugreifen, erstellen Sie eine Ansicht für die Tabellen der Staging- und Replikadatasets in BigQuery, in der die Zusammenführung nachgeahmt wird. Diese Ansicht wird als eine logische Tabelle erstellt (für das Staging- und das Replikadataset). Wenn die Zusammenführungsfrequenz niedrig ist und Sie schnelleren Zugriff auf die Daten benötigen, verwenden Sie die Ansicht.