Deepak Verma created BEAM-9417:
----------------------------------
Summary: Unable to Read form BigQuery and File system in same
pipeline
Key: BEAM-9417
URL: https://issues.apache.org/jira/browse/BEAM-9417
Project: Beam
Issue Type: Bug
Components: io-py-gcp
Environment: macbook pro cataline, python3.7, apache-beam[gcp]===2.19.0
Reporter: Deepak Verma
I am trying to read from Bigquery and Local file system in my apache beam[gcp]
pipeline.
{code:java}
pipeline_options = PipelineOptions()
options = pipeline_options.view_as(PreProcessOptions)
options.view_as(SetupOptions).save_main_session = True
p = beam.Pipeline(options=options)
apn_query = "select * from `{bq_project}.config.apn` where
customer='{customer}'"\
.format(bq_project=options.bq_project, customer=options.customer)
file_path = "mycsv.csv.gz"
apn = p | beam.io.Read(beam.io.BigQuerySource(query=apn_query,
use_standard_sql=True))
preprocess_rows = p | beam.io.ReadFromText(file_path, coder=UnicodeCoder())
{code}
When I am running this job, I am getting below error
{code:java}
Traceback (most recent call last):
File "/etl/dataflow/etlTXLPreprocessor.py", line 125, in <module>
run()
File "/etl/dataflow/etlTXLPreprocessor.py", line 120, in run
p.run().wait_until_finish()
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
461, in run
self._options).run(False)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
474, in run
return self.runner.run_pipeline(self, self._options)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py",
line 182, in run_pipeline
return runner.run_pipeline(pipeline, options)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py",
line 413, in run_pipeline
pipeline.replace_all(_get_transform_overrides(options))
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
443, in replace_all
self._replace(override)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
340, in _replace
self.visit(TransformUpdater(self))
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
503, in visit
self._root_transform().visit(visitor, self, visited)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
939, in visit
part.visit(visitor, pipeline, visited)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
939, in visit
part.visit(visitor, pipeline, visited)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
939, in visit
part.visit(visitor, pipeline, visited)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
942, in visit
visitor.visit_transform(self)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
338, in visit_transform
self._replace_if_needed(transform_node)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/pipeline.py", line
301, in _replace_if_needed
new_output = replacement_transform.expand(input_node)
File
"/etl/dataflow/venv3/lib/python3.7/site-packages/apache_beam/runners/direct/sdf_direct_runner.py",
line 87, in expand
invoker = DoFnInvoker.create_invoker(signature, process_invocation=False)
File "apache_beam/runners/common.py", line 360, in
apache_beam.runners.common.DoFnInvoker.create_invoker
TypeError: create_invoker() takes at least 2 positional arguments (1
given){code}
But If I run my code like this
{code:java}
pipeline_options = PipelineOptions()
options = pipeline_options.view_as(PreProcessOptions)
options.view_as(SetupOptions).save_main_session = True
p = beam.Pipeline(options=options)
file_path = "mycsv.csv.gz"
preprocess_rows = p | beam.io.ReadFromText(file_path, coder=UnicodeCoder())
{code}
or like this
{code:java}
pipeline_options = PipelineOptions()
options = pipeline_options.view_as(PreProcessOptions)
options.view_as(SetupOptions).save_main_session = True
p = beam.Pipeline(options=options)
apn_query = "select * from `{bq_project}.config.apn` where
customer='{customer}'"\
.format(bq_project=options.bq_project, customer=options.customer)
apn = p | beam.io.Read(beam.io.BigQuerySource(query=apn_query,
use_standard_sql=True))
{code}
or even like this
{code:java}
pipeline_options = PipelineOptions()
options = pipeline_options.view_as(PreProcessOptions)
options.view_as(SetupOptions).save_main_session = True
p = beam.Pipeline(options=options)
apn_query = "select * from `{bq_project}.config.apn` where
customer='{customer}'"\
.format(bq_project=options.bq_project, customer=options.customer)
apn = p | beam.io.Read(beam.io.BigQuerySource(query=apn_query,
use_standard_sql=True))
apn = p | beam.io.Read(beam.io.BigQuerySource(query=apn_query,
use_standard_sql=True)){code}
the code just works fine.
Is it a limitation of the apache beam to read from the same source?
If so, can we add this feature?
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
