You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ib...@apache.org on 2020/04/16 23:18:40 UTC
[beam] branch release-2.21.0 updated: [BEAM-9769] Ensuring JSON is
the default export format for BQ sink
This is an automated email from the ASF dual-hosted git repository.
ibzib pushed a commit to branch release-2.21.0
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/release-2.21.0 by this push:
new b551282 [BEAM-9769] Ensuring JSON is the default export format for BQ sink
new 6e6e779 Merge pull request #11435 from pabloem/cp-BEAM-9769
b551282 is described below
commit b551282c62dc2451ef75c7bbc6f1a573556de843
Author: Pablo Estrada <pa...@apache.org>
AuthorDate: Wed Apr 15 15:49:57 2020 -0700
[BEAM-9769] Ensuring JSON is the default export format for BQ sink
fixup
Adding documentation to CHANGES.md
---
CHANGES.md | 11 ++++++++++-
sdks/python/apache_beam/io/gcp/bigquery.py | 7 ++++---
sdks/python/apache_beam/io/gcp/bigquery_file_loads.py | 6 +++---
sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py | 4 ++--
sdks/python/apache_beam/io/gcp/bigquery_test.py | 4 +---
5 files changed, 20 insertions(+), 12 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 4684ece..f2edcfa 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -84,7 +84,17 @@ fully typed and can contain complex rows. *Remark: Schema aware is still experim
* Java SDK: The protobuf extension is fully schema aware and also includes protobuf option
conversion to beam schema options. *Remark: Schema aware is still experimental.*
([BEAM-9044](https://issues.apache.org/jira/browse/BEAM-9044))
+* Added ability to write to BigQuery via Avro file loads (Python) ([BEAM-8841](https://issues.apache.org/jira/browse/BEAM-8841))
+ By default, file loads will be done using JSON, but it is possible to
+ specify the temp_file_format parameter to perform file exports with AVRO.
+ AVRO-based file loads work by exporting Python types into Avro types, so
+ to switch to Avro-based loads, you will need to change your data types
+ from Json-compatible types (string-type dates and timestamp, long numeric
+ values as strings) into Python native types that are written to Avro
+ (Python's date, datetime types, decimal, etc). For more information
+ see https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-avro#avro_conversions.
+
## Breaking Changes
* Dataflow runner now requires the `--region` option to be set, unless a default value is set in the environment ([BEAM-9199](https://issues.apache.org/jira/browse/BEAM-9199)). See [here](https://cloud.google.com/dataflow/docs/concepts/regional-endpoints) for more details.
@@ -118,7 +128,6 @@ Schema Options, it will be removed in version `2.23.0`. ([BEAM-9704](https://iss
* Added a PTransform for inspection and deidentification of text using Google Cloud DLP. (Python) ([BEAM-9258](https://issues.apache.org/jira/browse/BEAM-9258))
* New AnnotateText PTransform that integrates Google Cloud Natural Language functionality (Python) ([BEAM-9248](https://issues.apache.org/jira/browse/BEAM-9248))
* _ReadFromBigQuery_ now supports value providers for the query string (Python) ([BEAM-9305](https://issues.apache.org/jira/browse/BEAM-9305))
-* Added ability to write to BigQuery via Avro file loads (Python) ([BEAM-8841](https://issues.apache.org/jira/browse/BEAM-8841))
* Direct runner for FnApi supports further parallelism (Python) ([BEAM-9228](https://issues.apache.org/jira/browse/BEAM-9228))
* Support for _@RequiresTimeSortedInput_ in Flink and Spark (Java) ([BEAM-8550](https://issues.apache.org/jira/browse/BEAM-8550))
diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py b/sdks/python/apache_beam/io/gcp/bigquery.py
index 62df4f6..ff1c29a 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery.py
@@ -1366,8 +1366,9 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider` that has a JSON string,
validate: Indicates whether to perform validation checks on
inputs. This parameter is primarily used for testing.
temp_file_format: The format to use for file loads into BigQuery. The
- options are NEWLINE_DELIMITED_JSON or AVRO, with AVRO being used
- by default. For advantages and limitations of the two formats, see
+ options are NEWLINE_DELIMITED_JSON or AVRO, with NEWLINE_DELIMITED_JSON
+ being used by default. For advantages and limitations of the two
+ formats, see
https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-avro
and
https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-json.
@@ -1394,7 +1395,7 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider` that has a JSON string,
self.triggering_frequency = triggering_frequency
self.insert_retry_strategy = insert_retry_strategy
self._validate = validate
- self._temp_file_format = temp_file_format or bigquery_tools.FileFormat.AVRO
+ self._temp_file_format = temp_file_format or bigquery_tools.FileFormat.JSON
self.additional_bq_parameters = additional_bq_parameters or {}
self.table_side_inputs = table_side_inputs or ()
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py b/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
index 8479e7b..91b5419 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_file_loads.py
@@ -204,7 +204,7 @@ class WriteRecordsToFile(beam.DoFn):
self.schema = schema
self.max_files_per_bundle = max_files_per_bundle
self.max_file_size = max_file_size
- self.file_format = file_format or bigquery_tools.FileFormat.AVRO
+ self.file_format = file_format or bigquery_tools.FileFormat.JSON
def display_data(self):
return {
@@ -276,7 +276,7 @@ class WriteGroupedRecordsToFile(beam.DoFn):
self, schema, max_file_size=_DEFAULT_MAX_FILE_SIZE, file_format=None):
self.schema = schema
self.max_file_size = max_file_size
- self.file_format = file_format or bigquery_tools.FileFormat.AVRO
+ self.file_format = file_format or bigquery_tools.FileFormat.JSON
def process(self, element, file_prefix, *schema_side_inputs):
destination = element[0]
@@ -645,7 +645,7 @@ class BigQueryBatchFileLoads(beam.PTransform):
self.test_client = test_client
self.schema = schema
- self._temp_file_format = temp_file_format or bigquery_tools.FileFormat.AVRO
+ self._temp_file_format = temp_file_format or bigquery_tools.FileFormat.JSON
# If we have multiple destinations, then we will have multiple load jobs,
# thus we will need temporary tables for atomicity.
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py b/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
index 29fdfa5..83bbf26 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py
@@ -176,7 +176,7 @@ class TestWriteRecordsToFile(_TestCaseWithTempDirCleanUp):
file length is very small, so only a couple records fit in each file.
"""
- fn = bqfl.WriteRecordsToFile(schema=_ELEMENTS_SCHEMA, max_file_size=300)
+ fn = bqfl.WriteRecordsToFile(schema=_ELEMENTS_SCHEMA, max_file_size=50)
self.tmpdir = self._new_tempdir()
def check_many_files(output_pcs):
@@ -306,7 +306,7 @@ class TestWriteGroupedRecordsToFile(_TestCaseWithTempDirCleanUp):
file length is very small, so only a couple records fit in each file.
"""
fn = bqfl.WriteGroupedRecordsToFile(
- schema=_ELEMENTS_SCHEMA, max_file_size=300)
+ schema=_ELEMENTS_SCHEMA, max_file_size=50)
self.tmpdir = self._new_tempdir()
def check_multiple_files(output_pc):
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_test.py b/sdks/python/apache_beam/io/gcp/bigquery_test.py
index 33cf483..47e7e30 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_test.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_test.py
@@ -840,8 +840,7 @@ class BigQueryStreamingInsertTransformIntegrationTests(unittest.TestCase):
str, '%s:%s' % (self.project, output_table_2)),
schema=beam.io.gcp.bigquery.SCHEMA_AUTODETECT,
additional_bq_parameters=lambda _: additional_bq_parameters,
- method='FILE_LOADS',
- temp_file_format=bigquery_tools.FileFormat.JSON))
+ method='FILE_LOADS'))
@attr('IT')
def test_multiple_destinations_transform(self):
@@ -1035,7 +1034,6 @@ class PubSubBigQueryIT(unittest.TestCase):
_ = mesages | WriteToBigQuery(
self.output_table,
schema=self.SCHEMA,
- temp_file_format=bigquery_tools.FileFormat.JSON,
method=method,
triggering_frequency=triggering_frequency)