diff --git a/sdks/python/apache_beam/examples/snippets/snippets.py b/sdks/python/apache_beam/examples/snippets/snippets.py index aabbb95db3a4..2f76960754e4 100644 --- a/sdks/python/apache_beam/examples/snippets/snippets.py +++ b/sdks/python/apache_beam/examples/snippets/snippets.py @@ -1181,6 +1181,39 @@ def model_bigqueryio( create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED) # [END model_bigqueryio_write] + # [START model_bigqueryio_write_with_storage_write_api] + quotes | beam.io.WriteToBigQuery( + table_spec, + schema=table_schema, + method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API) + # [END model_bigqueryio_write_with_storage_write_api] + + # [START model_bigqueryio_write_schema] + table_schema = { + 'fields': [ + { + "name": "request_ts", + "type": "TIMESTAMP", + "mode": "REQUIRED" + }, + { + "name": "user_name", + "type": "STRING", + "mode": "REQUIRED" + } + ] + } + # [END model_bigqueryio_write_schema] + + # [START model_bigqueryio_storage_write_api_with_frequency_and_multiple_streams] + # The SDK for Python does not support `withNumStorageWriteApiStreams` + quotes | beam.io.WriteToBigQuery( + table_spec, + schema=table_schema, + method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API, + triggering_frequency=5) + # [END model_bigqueryio_storage_write_api_with_frequency_and_multiple_streams] + # [START model_bigqueryio_write_dynamic_destinations] fictional_characters_view = beam.pvalue.AsDict( pipeline | 'CreateCharacters' >> beam.Create([('Yoda', True), diff --git a/website/www/site/content/en/documentation/io/built-in/google-bigquery.md b/website/www/site/content/en/documentation/io/built-in/google-bigquery.md index 61ff8c74e6dc..2bc3a5b5f3c7 100644 --- a/website/www/site/content/en/documentation/io/built-in/google-bigquery.md +++ b/website/www/site/content/en/documentation/io/built-in/google-bigquery.md @@ -795,7 +795,7 @@ BigQueryIO.writeTableRows() ); {{< /highlight >}} {{< highlight py >}} -# The SDK for Python does not support the BigQuery Storage API. +{{< code_sample "sdks/python/apache_beam/examples/snippets/snippets.py" model_bigqueryio_write_with_storage_write_api >}} {{< /highlight >}} If you want to change the behavior of BigQueryIO so that all the BigQuery sinks @@ -820,7 +820,7 @@ TableSchema schema = new TableSchema().setFields( .setMode("REQUIRED"))); {{< /highlight >}} {{< highlight py >}} -# The SDK for Python does not support the BigQuery Storage API. +{{< code_sample "sdks/python/apache_beam/examples/snippets/snippets.py" model_bigqueryio_write_schema >}} {{< /highlight >}} For streaming pipelines, you need to set two additional parameters: the number @@ -834,7 +834,7 @@ BigQueryIO.writeTableRows() ); {{< /highlight >}} {{< highlight py >}} -# The SDK for Python does not support the BigQuery Storage API. +{{< code_sample "sdks/python/apache_beam/examples/snippets/snippets.py" model_bigqueryio_storage_write_api_with_frequency_and_multiple_streams >}} {{< /highlight >}} The number of streams defines the parallelism of the BigQueryIO Write transform