Apache Beam窗口化和sharding BigQuery输出表

Question

我的用例很简单：从pub/sub订阅中读取事件日志，解析它们并保存到BigQuery中。由于预计事件的数量将显著增加，而且我使用的是无界数据源，所以我决定在BigQuery中配置sharding：根据事件数据的时间戳（在Beam文档中称为“事件时间”）将事件存储到每日表中。我的问题是，我是否需要在我的情况下配置窗口，或者我可以只保留默认配置，隐式地使用全局窗口？我之所以问这个问题，是因为我发现的大多数BigQuery sharding示例都假设使用窗口配置。但是在我的例子中，由于我没有使用任何分组操作作为groupbykey和combine，所以在没有任何窗口配置的情况下应该可以。或者有什么理由让我使用窗口，例如，它可能会影响BigQueryio的执行方式？

下面是我现在分叉的方式。

static class TableNamingFn implements SerializableFunction<ValueInSingleWindow<TableRow>, TableDestination> {
    @Override
    public TableDestination apply(ValueInSingleWindow<TableRow> input) {
        DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyyMMdd").withZone(DateTimeZone.UTC);

        TableReference reference = new TableReference();
        reference.setProjectId("test-project");
        reference.setDatasetId("event_log");

        DateTime timestamp = new DateTime(input.getValue().get("event_timestamp"), DateTimeZone.UTC);
        reference.setTableId("events_" + formatter.print(timestamp));
        return new TableDestination(reference, null);
    }
}

// And then
eventRows.apply("BigQueryWrite", BigQueryIO.writeTableRows()
        .to(new TableNamingFn())
        .withSchema(EventSchema)
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

Answer 1

您是否考虑过使用日期分区表来代替，这看起来像是试图按日期对表进行分段。您可以使用分区装饰器更新设置表id的位置，如下所示：

reference.setTableId("events$" + formatter.print(timestamp));

本文介绍了如何将BigQuery的分区表与Apache Beam一起使用。特别是，这段代码可能是您想要使用的:https://gist.githubusercontent.com/AlexvanBoxel/902099911d86b6827c8ea07f4e1437d4/raw/cc8246eb9b3219550379cfe7b3b7abca8fc77401/media_bq_tableref_partition