我正试图使用Google Dataflow将数据从一个BigQuery表传输到另一个：
import apache_beam as beam
from apache_beam.io.gcp.internal.clients import bigquery
from apache_beam.options.pipeline_options import PipelineOptions
import argparse
def parseArgs():
parser = argparse.ArgumentParser()
parser.add_argument(
'--experiment',
default='use_unsupported_python_version',
help='This does not seem to do anything.')
args, beam_args = parser.parse_known_args()
return beam_args
def beamer(rows=[]):
if len(rows) == 0:
return
project = 'myproject-474601'
gcs_temp_location = 'gs://my_temp_bucket/tmp'
gcs_staging_location = 'gs://my_temp_bucket/staging'
table_spec = bigquery.TableReference(
projectId=project,
datasetId='mydataset',
tableId='test')
beam_options = PipelineOptions(
parseArgs(), # This doesn't seem to work.
project=project,
runner='DataflowRunner',
job_name='unique-job-name',
temp_location=gcs_temp_location,
staging_location=gcs_staging_location,
use_unsupported_python_version=True, # This doesn't work either. :(
experiment='use_unsupported_python_version' # This also doesn't work.
)
with beam.Pipeline(options=beam_options) as p:
quotes = p | beam.Create(rows)
quotes | beam.io.WriteToBigQuery(
table_spec,
# custom_gcs_temp_location = gcs_temp_location, # Not needed?
method='FILE_LOADS',
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
return
if __name__ == '__main__':
beamer(rows=[{'id': 'ein', 'value': None, 'year': None, 'valueHistory': [{'year': 2021, 'amount': 900}]}])
但显然Dataflow不支持我的Python版本，因为我收到了这个错误：
Exception: Dataflow runner currently supports Python versions ['3.6', '3.7', '3.8'], got 3.9.7 (default, Sep 16 2021, 08:50:36)
[Clang 10.0.0 ].
To ignore this requirement and start a job using an unsupported version of Python interpreter, pass --experiment use_unsupported_python_version pipeline option.
因此，我尝试向PipelineOptions添加一个use_unsupported_python_version参数，但没有成功。我还尝试了experiment选项。在官方的管道选项文档中，它显示了args被成功地合并到PipelineOptions中，所以我也尝试过。
然而，我继续得到相同的unsupported version错误。如何让Dataflow使用我的Python版本？
尝试传递experiments=['use_unsupported_python_version']。您也可以删除parseArgs的实现。
'--实验=use_unsupported_python_version'请添加类似上述的选项