增量加载和 BigQuery

Question

我正在编写一个增量加载管道，将数据从MySQL加载到BigQuery，并使用Google Cloud Datastore作为元数据存储库。

我当前的管道是这样写的：

PCollection<TableRow> tbRows = 
pipeline.apply("Read from MySQL",
        JdbcIO.<TableRow>read().withDataSourceConfiguration(JdbcIO.DataSourceConfiguration
                .create("com.mysql.cj.jdbc.Driver", connectionConfig)
                .withUsername(username)
                .withPassword(password)
                .withQuery(query).withCoder(TableRowJsonCoder.of())
                .withRowMapper(JdbcConverters.getResultSetToTableRow())))
    .setCoder(NullableCoder.of(TableRowJsonCoder.of()));

tbRows.apply("Write to BigQuery",
            BigQueryIO.writeTableRows().withoutValidation()
                    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND).to(outputTable));

tbRows.apply("Getting timestamp column",
                MapElements.into(TypeDescriptors.strings())
                        .via((final TableRow row) -> (String) row.get(fieldName)))
                .setCoder(NullableCoder.of(StringUtf8Coder.of())).apply("Max", Max.globally())
                .apply("Updating Datastore", ParDo.of(new DoFn<String, String>() {
                    @ProcessElement
                    public void processElement(final ProcessContext c) {
                        DatastoreConnector.udpate(table, c.element());
                    }
                }));

我面临的问题是，当BigQuery写入步骤失败时，数据存储仍在更新，是否有任何方法在更新数据存储之前等待BigQuery写完成？

谢谢！

Answer 1

目前，这不能在与BigQueryIO.writeTableRows（）的同一管道中完成，因为它会产生终端输出（PDone）。不过我有一些建议。

• 我怀疑BigQuery写入失败是罕见的。在这种情况下，您可以从辅助作业/进程中删除相应的Datastore数据。
• 您是否考虑过更适合写入增量更改数据的CDC解决方案。例如，请参阅此处的数据流模板。