Google数据流在BigQuery中写入多行

Question

我有一个简单的流程，目的是在一个BigQuery表中写两行。我使用动态目标，因为之后我将在多个表上写，在那个例子中是同一个表...问题是我的BigQuery表最后只有一行。在第二次插入时，我看到以下错误

"状态：{code： 6
消息："已存在：作业sampleProject et3：b9912b9b05794aec8f4292b2ae493612_eeb0082ade6f4a58a14753d1cc92ddbc_00001-0"
}"

这是什么意思？和这个限制有关吗？https://github . com/GoogleCloudPlatform/DataflowJavaSDK/issues/550我该如何做这项工作？

我用的是BeamSDK 2.0.0，我试过2.1.0(结果相同)

我发射的方式:

mvn compile exec:java-Dexec.mainClass=fr.gireve.dataflow。LogsFlowBug-Dexec。args=“--runner=DataflowRunner--inputDir=gs://sampleprojet3.appspot.com/--project=sampleprojet3--stagingLocation=gs://dataflow-sampleprojet3/tmp”-Pdataflow runner

    Pipeline p = Pipeline.create(options);

    final List<String> tableNameTableValue = Arrays.asList("table1:value1", "table1:value2", "table2:value1", "table2:value2");

    p.apply(Create.of(tableNameTableValue)).setCoder(StringUtf8Coder.of())
            .apply(BigQueryIO.<String>write()
            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
            .to(new DynamicDestinations<String, KV<String, String>>() {
                @Override
                public KV<String, String> getDestination(ValueInSingleWindow<String> element) {
                    final String[] split = element.getValue().split(":");
                    return KV.of(split[0], split[1]) ;
                }

                @Override
                public Coder<KV<String, String>> getDestinationCoder() {
                    return KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());
                }

                @Override
                public TableDestination getTable(KV<String, String> row) {
                    String tableName = row.getKey();
                    String tableSpec = "sampleprojet3:testLoadJSON." + tableName;
                    return new TableDestination(tableSpec, "Table " + tableName);
                }

                @Override
                public TableSchema getSchema(KV<String, String> row) {
                    List<TableFieldSchema> fields = new ArrayList<>();
                    fields.add(new TableFieldSchema().setName("myColumn").setType("STRING"));
                    TableSchema ts = new TableSchema();
                    ts.setFields(fields);
                    return ts;
                }
            })
            .withFormatFunction(new SerializableFunction<String, TableRow>() {
                public TableRow apply(String row) {
                    TableRow tr = new TableRow();
                    tr.set("myColumn", row);
                    return tr;
                }
            }));

    p.run().waitUntilFinish();

谢谢

Answer 1

DynamicDestments将每个元素与一个目标相关联-即元素应该去的地方。元素根据其目的地路由到BigQuery表：1个目的地=1个带有架构的BigQuery表：目的地应该包含足够的信息来生成TableDestation和架构。具有相同目的地的元素转到相同的表，具有不同目的地的元素转到不同的表。

您的代码片段使用动态目的地，其目标类型包含元素和表，这是不必要的，当然，违反了上述约束：具有不同目标的元素最终转到同一个表：例如，KV（“table1”，“value1”）和KV（“table1”，“value2”）是不同的目标，但您的getTable将它们映射到同一个表表1。

您需要从目标类型中删除该元素。这也将导致更简单的代码。作为旁注，我认为你不需要重写 getDestinationCoder（） - 它可以自动推断出来。

试试这个:

        .to(new DynamicDestinations<String, String>() {
            @Override
            public String getDestination(ValueInSingleWindow<String> element) {
                return element.getValue().split(":")[0];
            }

            @Override
            public TableDestination getTable(String tableName) {
                return new TableDestination(
                    "sampleprojet3:testLoadJSON." + tableName, "Table " + tableName);
            }

            @Override
            public TableSchema getSchema(String tableName) {
                List<TableFieldSchema> fields = Arrays.asList(
                    new TableFieldSchema().setName("myColumn").setType("STRING"));
                return new TableSchema().setFields(fields);
            }
        })