ahmedabu98 commented on issue #22543:
URL: https://github.com/apache/beam/issues/22543#issuecomment-1208560215
With the changes in #22624 I was able to write successfully with the
following pipeline:
```
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import java.util.Arrays;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import
org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryOptions;
import org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinations;
import org.apache.beam.sdk.io.gcp.bigquery.TableDestination;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.checkerframework.checker.nullness.qual.Nullable;
public class UpdateSchemaDestinationWrite {
static final String PROJECT_AND_DATASET_ID = "<myproject>:<mydataset>.";
private static final class SomeDynamicDestinations
extends DynamicDestinations<KV<String, String>, String> {
private static final long serialVersionUID = 1L;
@Override
public String getDestination(@Nullable ValueInSingleWindow<KV<String,
String>> element) {
return element.getValue().getKey();
}
@Override
public TableDestination getTable(String destination) {
return new TableDestination(PROJECT_AND_DATASET_ID + destination, "a
table");
}
@Override
public @Nullable TableSchema getSchema(String destination) {
return new TableSchema()
.setFields(Arrays.asList(new
TableFieldSchema().setName("name").setType("STRING")));
}
}
public static void main(String args[]) {
BigQueryOptions bqOptions =
PipelineOptionsFactory.fromArgs(args).as(BigQueryOptions.class);
Pipeline pipeline = Pipeline.create(bqOptions);
DynamicDestinations<KV<String, String>, String> destinations = new
SomeDynamicDestinations();
PCollection<KV<String, String>> rows =
pipeline.apply(Create.of(KV.of("my_table", "hi"), KV.of("my_table",
"hello")));
rows.apply(
BigQueryIO.<KV<String, String>>write()
.withFormatFunction(kv -> new TableRow().set("name",
kv.getValue()))
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(WriteDisposition.WRITE_APPEND)
.withMaxBytesPerPartition(1)
.to(destinations));
pipeline.run().waitUntilFinish();
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
