为什么使用 Dataflow 写入 Bigquery 非常慢？

Question

我可以以每秒大约 10,000 次插入的速度将插入直接流式传输到 BigQuery，但是当我尝试使用 Dataflow 插入时，'ToBqRow' 步骤（如下所示）非常慢。 每 10 分钟仅 50 行，这是 4 名工人。知道为什么吗？相关代码如下：

PCollection<Status> statuses = p
        .apply("GetTweets", PubsubIO.readStrings().fromTopic(topic))
        .apply("ExtractData", ParDo.of(new DoFn<String, Status>() {
    @ProcessElement
    public void processElement(DoFn<String, Status>.ProcessContext c) throws Exception {
            String rowJson = c.element();

            try {
                TweetsWriter.LOGGER.debug("ROWJSON = " + rowJson);
                Status status = TwitterObjectFactory.createStatus(rowJson);
                if (status == null) {
                    TweetsWriter.LOGGER.error("Status is null");
                } else {
                    TweetsWriter.LOGGER.debug("Status value: " + status.getText());
                }
                c.output(status);
                TweetsWriter.LOGGER.debug("Status: " + status.getId());
            } catch (Exception var4) {
                TweetsWriter.LOGGER.error("Status creation from JSON failed: " + var4.getMessage());
            }

    }
}));

statuses
        .apply("ToBQRow", ParDo.of(new DoFn<Status, TableRow>() {
            @ProcessElement
            public void processElement(ProcessContext c) throws Exception {
                TableRow row = new TableRow();
                Status status = c.element();
                row.set("Id", status.getId());
                row.set("Text", status.getText());
                row.set("RetweetCount", status.getRetweetCount());
                row.set("FavoriteCount", status.getFavoriteCount());
                row.set("Language", status.getLang());
                row.set("ReceivedAt", (Object)null);
                row.set("UserId", status.getUser().getId());
                row.set("CountryCode", status.getPlace().getCountryCode());
                row.set("Country", status.getPlace().getCountry());
                c.output(row);
        }
    }))
        .apply("WriteTableRows", BigQueryIO.writeTableRows().to(tweetsTable)
                .withSchema(schema)
                .withMethod(Method.STREAMING_INSERTS)
                .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED));

p.run();

Answer 1

事实证明 Dataflow 下的 Bigquery 并不慢。问题是，'status.getPlace().getCountryCode()' 返回 NULL 所以它抛出 NullPointerException，我在日志中的任何地方都看不到！显然，Dataflow 日志记录需要改进。现在运行真的很好。一旦消息进入主题，几乎立即它就会写入 BigQuery！

为什么使用 Dataflow 写入 Bigquery 非常慢？

Why is writing to Bigquery using Dataflow EXTREMELY slow?

google-bigquery

google-cloud-platform

google-cloud-dataflow

apache-beam