claudevdm commented on code in PR #39236:
URL: https://github.com/apache/beam/pull/39236#discussion_r3725366522
##########
sdks/python/apache_beam/io/gcp/bigquery.py:
##########
@@ -197,6 +197,43 @@ def compute_table_name(row):
a tuple of PCollectionViews to be passed to the schema callable (much like
the `table_side_inputs` parameter).
+Dynamic Schemas with Storage Write API
+--------------------------------------
+When writing to dynamic destinations with `method=STORAGE_WRITE_API`, a union
schema
+containing all fields across destination tables is required at the PCollection
level
Review Comment:
Is the expectation that the destination schema for every table will be the
full union schema, or each table is meant to have the individually mapped
schema and union is just a trick to make it work? Can you add an integration
test that validates the expected schema on the tables and document the expected
behavior?
##########
.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Dataflow.json:
##########
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run",
- "modification": 16
+ "modification": 21
Review Comment:
I think you can also trigger
.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json for faster
validation
##########
sdks/python/apache_beam/io/external/xlang_bigqueryio_it_test.py:
##########
@@ -483,6 +484,74 @@ def test_write_to_dynamic_destinations(self):
use_at_least_once=False))
hamcrest_assert(p, all_of(*bq_matchers))
+ def test_write_to_dynamic_destinations_with_dynamic_schema(self):
Review Comment:
Can we add a test for the case where the destination tables are pre-created/
already exist beforehand?
##########
sdks/python/apache_beam/io/external/xlang_bigqueryio_it_test.py:
##########
@@ -483,6 +484,74 @@ def test_write_to_dynamic_destinations(self):
use_at_least_once=False))
hamcrest_assert(p, all_of(*bq_matchers))
+ def test_write_to_dynamic_destinations_with_dynamic_schema(self):
+ base_table_spec = '{}.dynamic_dest_dyn_schema_'.format(self.dataset_id)
+ spec_with_project = '{}:{}'.format(self.project, base_table_spec)
+ table_a = base_table_spec + 'users'
+ table_b = base_table_spec + 'scores'
+
+ schema_a = "id:INTEGER,name:STRING"
+ schema_b = "id:INTEGER,score:INTEGER,active:BOOLEAN"
+
+ elements_a = [
+ {
+ 'id': 1, 'name': 'alice'
+ },
+ {
+ 'id': 2, 'name': 'bob'
+ },
+ ]
+ elements_b = [
+ {
+ 'id': 101, 'score': 95, 'active': True
+ },
+ {
+ 'id': 102, 'score': 80, 'active': False
+ },
+ ]
+ elements = elements_a + elements_b
+
+ schema_map = {
+ spec_with_project + 'users': schema_a,
+ spec_with_project + 'scores': schema_b,
+ }
+
+ bq_matchers = [
+ BigqueryFullResultMatcher(
+ project=self.project,
+ query="SELECT id, name FROM %s" % table_a,
+ data=self.parse_expected_data(elements_a)),
+ BigqueryFullResultMatcher(
+ project=self.project,
+ query="SELECT id, score, active FROM %s" % table_b,
+ data=self.parse_expected_data(elements_b)),
+ ]
+
+ def get_destination(record):
+ if 'name' in record:
+ return spec_with_project + 'users'
+ return spec_with_project + 'scores'
+
+ def get_schema_raw(dest, side_map):
+ return side_map[dest]
+
+ get_schema = bigquery.dynamic_schema(
+ get_schema_raw,
+ union_schema="id:INTEGER,name:STRING,score:INTEGER,active:BOOLEAN")
+
+ with beam.Pipeline(argv=self.args) as p:
+ schema_pc = p | "CreateSchema" >> beam.Create([schema_map])
+ _ = (
+ p
+ | "CreateElements" >> beam.Create(elements)
+ | beam.io.WriteToBigQuery(
+ table=get_destination,
+ method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API,
+ schema=get_schema,
+ schema_side_inputs=(beam.pvalue.AsSingleton(schema_pc), ),
+ use_at_least_once=False))
+ hamcrest_assert(p, all_of(*bq_matchers))
+
Review Comment:
Can we also validate the expected schema of the tables that were created
after the pipeline ran?
--
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]
