如何暂存 GCP/Apache Beam 数据流模板? [英] How to I stage a GCP/Apache Beam Dataflow template?
问题描述
好的,我必须在这里遗漏一些东西.将管道作为模板进行暂存需要什么?当我尝试通过 这些说明,它运行模块但不进行任何操作.它似乎按预期运行而没有错误,但我没有看到任何文件实际上被添加到存储桶位置,在我的 --template_location 中侦听.我的python代码应该出现在那里吗?我猜对吗?我已经确定我已经安装了所有的 Beam 和谷歌云 SDK,但也许我遗漏了什么?您需要做什么来暂存此数据流模板?我也可以手动将文件放入存储桶并从那里运行吗?以下是我目前正在使用的模板:
Ok I have to be missing something here. What do i need to stage a pipeline as a template? When I try to stage my template with via these instructions, it runs the module but doesn't stage anything., it appears to function as expected without errors, but I don't see any files actually get added to the bucket location listen in my --template_location. Should my python code be showing up there? I assume so right? I have made sure i have all the beam and google cloud SDKs installed, but maybe I'm missing something? What do you need to do to stage this dataflow template? Also can I manually just drop the file in a bucket and run it from there? The following is the template I am currently playing with:
import json
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json
GC_PROJECT = 'my-proj'
BUCKET = 'test-bucket'
STAGING_BUCKET = '%s/test' % BUCKET
TEMP_BUCKET = '%s/test' % BUCKET
# RUNNER = 'DataflowRunner'
RUNNER = 'DirectRunner'
# pipeline_args = ['--save_main_session']
pipeline_args = []
pipeline_args.append('--project=%s' % GC_PROJECT)
pipeline_args.append('--runner=%s' % RUNNER)
pipeline_args.append('--staging_location=gs://%s' % STAGING_BUCKET)
pipeline_args.append('--temp_location=gs://%s' % TEMP_BUCKET)
BQ_DATASET = 'lake'
BQ_TABLE = 'whatever'
SCHEMA_OBJ = [
{"name": "id", "type": "STRING", "description": ""},
{"name": "value", "type": "STRING", "description": ""}
]
class ContactUploadOptions(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
parser.add_value_provider_argument(
'--infile',
type=str,
help='path of input file',
default='gs://%s/data_files/test.csv' % BUCKET)
def run(argv=None):
print('running')
p = beam.Pipeline(options=PipelineOptions(pipeline_args))
lines = (p
| beam.Create([
{"id": "some random name", "value": "i dont know"},
{"id": "id2", "value": "whatever man"}]))
schema_str = '{"fields": ' + json.dumps(SCHEMA_OBJ) + '}'
schema = parse_table_schema_from_json(schema_str)
output_destination = '%s.%s' % (BQ_DATASET, BQ_TABLE)
(lines
| 'Write lines to BigQuery' >> beam.io.WriteToBigQuery(
output_destination,
schema=schema,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
p.run().wait_until_finish()
if __name__ == '__main__':
run(pipeline_args)
此外,如果有人可以链接一些 sdk 文档/资源来解释上面的暂存说明如何/为什么应该起作用,那就太棒了!
Also, if someone could link some sdk documentaion/resources that explain how/why the staging instructions above are supposed to work, that would be awesome!
推荐答案
临时位置是在运行作业时加载临时文件的位置.您没有提到将在其中创建模板的template_location".
The temp location is where the temporary files will be loaded while running the job. You have not mentioned the "template_location" where template will be created.
这篇关于如何暂存 GCP/Apache Beam 数据流模板?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!