Google Dataflow write multiple line in BigQuery

Question

I have a simple flow which aim is to write two lines in one BigQuery Table. I use a DynamicDestinations because after that I will write on mutliple Table, on that example it's the same table... The problem is that I only have 1 line in my BigQuery table at the end. It stacktrace I see the following error on the second insert

" status: { code: 6
message: "Already Exists: Job sampleprojet3:b9912b9b05794aec8f4292b2ae493612_eeb0082ade6f4a58a14753d1cc92ddbc_00001-0"
} "

What does it means ? Is it related to this limitation ? https://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/550 How can I do the job ?

I use BeamSDK 2.0.0, I have try with 2.1.0 (same result)

The way I launch :

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();

Thanks

DynamicDestinations.getTable() must return different tables for different destinations, but this code ignores the .getValue() component of your KV: can you explain why you're using KV as a destination type, rather than just using table name as a destination type, since the value of the KV is ignored anyway? — jkff
Ok I agree it's not clear. I have updates the code. The input list present a set of string where first token represent the table. So at the end we must have two table created with 2 lines on each. Unfortunatly we have 2 tables but one lines :( — Stéphane Loison

jkff jkff · Accepted Answer · 2017-09-06T00:16:18

DynamicDestinations associates each element with a destination - i.e. where the element should go. Elements are routed to BigQuery tables according to their destinations: 1 destination = 1 BigQuery table with a schema: the destination should include just enough information to produce a TableDestination and a schema. Elements with the same destination go to the same table, elements with different destinations go to different tables.

Your code snippet uses DynamicDestinations with a destination type that contains both the element and the table, which is unnecessary, and of course, violates the constraint above: elements with a different destination end up going to the same table: e.g. KV("table1", "value1") and KV("table1", "value2") are different destinations but your getTable maps them to the same table table1.

You need to remove the element from your destination type. That will also lead to simpler code. As a side note, I think you don't need to override getDestinationCoder() - it can be inferred automatically.

Try this:

        .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);
            }
        })

Google Dataflow write multiple line in BigQuery

1 Answers