Stream-zu-Stream-Joins in kontinuierlichen Abfragen verwenden

In BigQuery können Sie Stream-zu-Stream-Joins in kontinuierlichen Abfragen verwenden, um Daten aus zwei oder mehr Echtzeit-Datenstreams zu analysieren und zu korrelieren. Stream-zu-Stream-Joins sind ein Join-Vorgang zwischen zwei oder mehr Tabellen, die zeitbezogene Daten aufnehmen.

Häufige Anwendungsfälle für Stream-zu-Stream-Joins sind die Erkennung von Finanzbetrug, die Erstellung von Kundenprofilen und die Optimierung des Lieferkettenmanagements. Diese Joins sind ein wichtiger Typ von zustandsorientierten Vorgängen. Weitere Informationen zu zustandsorientierten Vorgängen finden Sie unter Unterstützte zustandsorientierte Vorgänge.

Unterstützte JOIN-Typen

Die folgenden JOIN-Typen werden von kontinuierlichen Abfragen unterstützt:

  • Stream-zu-Stream-JOIN: Ein Join-Vorgang zwischen zwei oder mehr Tabellen, die zeitorientierte Datenaufnahme erhalten.
  • INNER JOIN

Nicht unterstützte JOIN-Typen

Die folgenden JOIN-Typen werden von kontinuierlichen Abfragen nicht unterstützt:

  • Stream-zu-statische-Tabelle-JOINs: Ein JOIN, bei dem mindestens eine verknüpfte Tabelle eine statische Tabelle ist, die keine zeitorientierte Datenaufnahme empfängt. Ein Beispiel für eine statische Tabelle ist eine Dimensionstabelle.
  • Andere Formen von JOIN-Vorgängen außer INNER.
  • JOINs, die keine zeitorientierten JOIN-Klauseln haben.

Daten aus mehreren Streams zusammenführen

Die folgende Abfrage zeigt, wie Sie eine Tabelle mit Taxifahrten mit einer Tabelle mit Taxianfragen verknüpfen und das nächstgelegene verfügbare Taxi für den Anfragenden innerhalb eines Zeitfensters von 5 Minuten ermitteln und diese Daten in eine andere BigQuery-Tabelle exportieren:

INSERT INTO
 `real_time_taxi_streaming.matched_rides`
SELECT
 requests.timestamp AS request_time,
 requests.request_id,
 taxis.taxi_id,
 ST_DISTANCE(
   ST_GEOGPOINT(requests.longitude, requests.latitude),
   ST_GEOGPOINT(taxis.longitude, taxis.latitude)
   ) AS distance_in_meters,
 taxis.timestamp AS taxi_available_time
FROM
 APPENDS (TABLE `real_time_taxi_streaming.ride_requests`,
   CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS requests
INNER JOIN
 APPENDS (TABLE `real_time_taxi_streaming.taxirides`,
   CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS taxis
ON
 requests.geohash = taxis.geohash
WHERE
 taxis.ride_status = 'available'
 AND taxis._CHANGE_TIMESTAMP BETWEEN(requests._CHANGE_TIMESTAMP - INTERVAL 5 MINUTE) AND requests._CHANGE_TIMESTAMP
 AND ST_DWITHIN(
   ST_GEOGPOINT(requests.longitude, requests.latitude),
   ST_GEOGPOINT(taxis.longitude, taxis.latitude),
   2000 -- Distance in meters
   );

Überlegungen zu Joins

In den folgenden Abschnitten werden die erforderlichen Überlegungen beim Ausführen eines Stream-zu-Stream-Joins beschrieben.

Beschränkungen

  • Es werden nur INNER JOIN-Vorgänge unterstützt. Andere Formen wie LEFT oder FULL OUTER werden nicht unterstützt.
  • Für jede Seite des INNER JOIN-Vorgangs muss eine Startzeit für die kontinuierliche Abfrage angegeben werden.
  • Zusätzlich zu einem Join-Schlüssel (z. B. table1.user_id = table2.user_id) muss die JOIN-Klausel eine Bedingung enthalten, um die Zeitstempelspalte auf ein konstantes Intervall zu beschränken. Diese Bedingung begrenzt, wie lange das System auf ein übereinstimmendes Ereignis im anderen Stream wartet. Sie können beispielsweise festlegen, dass ein Ereignis aus einem Stream nur dann mit einem Ereignis aus einem anderen Stream zusammengeführt werden kann, wenn die Zeitstempel innerhalb eines Intervalls von 5 Minuten liegen. Sie sind nicht auf symmetrische Intervalle beschränkt. Sie können beispielsweise ein 5-Minuten-Intervall auf einer Seite der Join-Bedingung und ein 1-Stunden-Intervall auf der anderen Seite verwenden.
  • Wenn Sie eine kontinuierliche Abfrage mit einem Stream-zu-Stream-Join starten, wird nur die Funktion APPENDS unterstützt. Die Funktion CHANGES wird nicht unterstützt.
  • Die BigQuery-Systemzeitspalte, die durch _CHANGE_TIMESTAMP definiert wird, ist die einzige unterstützte Zeitstempelspalte für Join-Vorgänge. Benutzerdefinierte Spalten werden nicht unterstützt.
  • Bei Verwendung in Verbindung mit Fensteraggregationen müssen Sie alle dokumentierten Einschränkungen für Fensteraggregationen beachten.

Kosten

Kontinuierliche BigQuery-Abfragen werden basierend auf der Rechenkapazität (Slots) abgerechnet, die während der Ausführung des Jobs verbraucht wird. Dieses rechenbasierte Modell gilt auch für zustandsorientierte Vorgänge wie Joins. Da Joins erfordern, dass das System den „Status“ speichert, während die Abfrage aktiv ist, verbrauchen sie zusätzliche Slot-Ressourcen. Wenn mehr Kontext oder Daten in einem Join gespeichert werden, z. B. wenn Sie längere Zeitintervalle in der JOIN- oder WHERE-Klausel verwenden, muss mehr Status beibehalten werden, was zu einer höheren Slot-Auslastung führt.

Nächste Schritte