我有一个数据流作业，我将首先从 bigquery 查询中读取（在标准 sql 中）。它在直接运行模式下完美运行。但是，我尝试在数据流运行程序模式下运行此数据流并遇到此错误：
响应： , content
显然 use_standard_sql 参数在数据流运行程序模式下不起作用。
版本：
阿帕奇光束：2.24.0
蟒蛇：3.8
last_update_date = pipeline | 'Read last update date' >> beam.io.Read(beam.io.BigQuerySource(
query='''
SELECT
MAX(date) AS date
FROM
GoogleSearchConsole.search_query_analytics_log
''',
use_standard_sql=True
))
尝试以下从 Bigquery 读取数据并将数据写入 Bigquery 的代码。
代码是 apache beam 数据流运行程序代码：-
#------------Import Lib-----------------------#
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
import apache_beam as beam, os, sys, argparse, logging
from apache_beam.options.pipeline_options import SetupOptions
#------------Set up BQ parameters-----------------------#
# Replace with Project Id
project = 'xxxxx'
#plitting Of Records----------------------#
def run(argv=None, save_main_session=True):
parser = argparse.ArgumentParser()
parser.add_argument(
'--cur_suffix',
dest='cur_suffix',
help='Input table suffix to process.')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
p1 = beam.Pipeline(options=pipeline_options)
logging.info('***********')
logging.info(known_args.cur_suffix)
data_loading = (
p1
| 'ReadFromBQ' >> beam.io.Read(beam.io.BigQuerySource(query='''SELECT SUBSTR(_time, 1, 19) as _time, dest FROM `project.dataset.table`''', use_standard_sql=True))
)
project_id = "xxxxxxx"
dataset_id = 'AAAAAAAA'
table_schema_Audit = ('_time:DATETIME, dest:STRING')
#---------------------Type = audit----------------------------------------------------------------------------------------------------------------------
result = (
data_loading
| 'Write-Audit' >> beam.io.WriteToBigQuery(
table='YYYYYYY',
dataset=dataset_id,
project=project_id,
schema=table_schema_Audit,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE
))
result = p1.run()
result.wait_until_finish()
if __name__ == '__main__':
path_service_account = 'ABGFDfc927.json'
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = path_service_account
run()
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
