在持续查询中使用流到流联接

在 BigQuery 中，您可以在持续查询中使用流到流联接，以分析和关联来自两个或更多实时数据流的数据。流到流联接是指在两个或多个接收面向时间的数据注入的表之间执行的联接操作。

流到流联接的常见应用场景包括金融欺诈检测、构建客户资料和优化供应链管理。这些联接是一种关键的有状态操作。如需详细了解有状态操作，请参阅支持的有状态操作。

支持的 JOIN 类型

持续查询支持以下 JOIN 类型：

  • 流到流 JOIN - 两个或更多接收面向时间的数据注入的表之间的联接操作。
  • INNER JOIN。

不支持的 JOIN 类型

持续查询不支持以下 JOIN 类型：

  • 流到静态表联接 - 至少有一个联接表是静态表的联接，不接收面向时间的数据注入。静态表的一个示例是维度表。
  • 除 INNER 之外的其他形式的 JOIN 运算。
  • 不包含时间导向型 JOIN 子句的 JOIN。

联接来自多个数据流的数据

以下查询展示了如何将出租车行程表与出租车请求表联接，并在 5 分钟的时间窗口内识别距离请求者最近的可用出租车，然后将这些数据导出到另一个 BigQuery 表中：

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
   );

联接注意事项

以下部分介绍了执行流到流联接时需要注意的事项。

限制

  • 仅支持 INNER JOIN 操作；不支持其他形式，例如 LEFT 或 FULL OUTER。
  • INNER JOIN 操作的每一侧都必须指定持续查询的开始时间。
  • 除了联接键（例如 table1.user_id = table2.user_id）之外，JOIN 子句还必须包含一个条件，用于将时间戳列限制为常量间隔。此条件限制了系统等待匹配事件到达另一个数据流的时长。例如，您可以指定，只有当一个流中的事件与另一个流中的事件的时间戳在 5 分钟间隔内时，才能将这两个事件联接。您不必局限于对称区间。例如，您可以在联接条件的一侧使用 5 分钟的时间间隔，在另一侧使用 1 小时的时间间隔。
  • 使用流到流联接启动持续查询时，仅支持 APPENDS 函数。不支持 CHANGES 函数。
  • 由 _CHANGE_TIMESTAMP 定义的 BigQuery 系统时间列是唯一支持联接操作的时间戳列。不支持用户定义的列。
  • 与窗口化聚合结合使用时，您必须遵循所有记录的窗口聚合限制。

价格注意事项

BigQuery 持续查询根据作业运行时消耗的计算容量（槽）计费。这种基于计算的模型也适用于联接等有状态操作。由于联接要求系统在查询处于活跃状态时存储“状态”，因此它们会消耗额外的槽资源。在联接中存储的上下文或数据越多（例如，在 JOIN 或 WHERE 子句中使用较长的时间间隔），需要保留的状态就越多，从而导致槽利用率越高。

后续步骤