从 Java 在 BigQuery 中创建数据加载表

Question

根据此处的文档：https://cloud.google.com/bigquery/docs/tables#creating_a_table_when_you_load_data BigQuery 应该可以根据数据创建表。

将数据加载到 BigQuery 时，您可以将数据加载到新表或分区中，可以将数据附加到现有表或分区中，也可以覆盖表或分区。在将数据加载到其中之前，您不需要创建一个空表。您可以同时创建新表并加载数据。

但是，每当我尝试将数据从 Java 流式传输到 BigQuery 时，我都会收到一条错误消息。

这是一个插入语句的示例，但仅在我手动创建表之后才有效：

InsertAllResponse response = bigQuery
        .insertAll(
                InsertAllRequest
                        .newBuilder(tableId)
                        .addRow(rowContent)
                        .build()
        );

我可以在 Java 中创建架构，然后创建表，但是我必须不断检查架构是否已创建，然后才能流式传输到它。 generateBigQuerySchema 是我创建的定义模式的方法。如果架构已经存在，下面的代码将失败，所以我必须在创建它之前检查它是否存在。

InsertAllResponse response = bigQuery
        .create(requestLog.generateBigQuerySchema(tableId))
        .getBigQuery()
        .insertAll(
                InsertAllRequest
                        .newBuilder(tableId)
                        .addRow(rowContent)
                        .build()
        );

Answer 1

我认为您根据API Reference 混合了两种不同的资源类型。 我的意思是Jobs 和Tabledata。

Jobs does loading where insertAll 方法来自 Tabledata，doesn't：

一次将数据流式传输到 BigQuery 一条记录，而无需运行 加载作业

我发现 Google 文档可能会像上面那样被误解，因为 Introduction to Loading Data into BigQuery 引用了流式插入 (insertAll)。如下所示：

你可以加载数据：

...通过使用streaming inserts 插入单个记录...

流式插入重定向到 Streaming Data into BigQuery 的位置，它讲述的是流式传输而不是加载：

您可以选择将数据加载到 BigQuery 中，而不是使用作业 使用 tabledata().insertAll() 方法。

关于streaming inserts (insertAll)的最后一件事：

确保您对数据集具有写入权限 包含您的目标表。 表格必须在您开始之前存在 除非您使用模板表，否则将数据写入其中。更多 有关模板表的信息，请参阅自动创建表 使用模板表。

如果您仍想同时使用模板表加载而不是流式传输并创建表，请使用 Jobs 和 load type of job（或其他类型，如果需要）

来自我的question的示例代码：

Insert insert = bigquery.jobs().insert(projectId,
                   new Job().setConfiguration(
                            new JobConfiguration().setLoad(
                                   new JobConfigurationLoad()
                                                .setSourceFormat("NEWLINE_DELIMITED_JSON")
                                                .setDestinationTable(
                                                        new TableReference()
                                                                .setProjectId(projectId)
                                                                .setDatasetId(dataSetId)
                                                                .setTableId(tableId)
                                                )
                                                .setCreateDisposition("CREATE_IF_NEEDED")
                                                .setWriteDisposition(writeDisposition)
                                                .setSourceUris(Collections.singletonList(sourceUri))
                                                .setAutodetect(true)
                                )
                        ));

Job myInsertJob = insert.execute();