亚洲在线久爱草,狠狠天天香蕉网,天天搞日日干久草,伊人亚洲日本欧美

為了賬號安全,請及時綁定郵箱和手機立即綁定
已解決430363個問題,去搜搜看,總會有你想問的

如何使用apache beam python在管道中附加結果?

如何使用apache beam python在管道中附加結果?

慕妹3146593 2021-12-08 10:15:47
我有 apache 光束管道,我使用 pubsub 從輸入文件中獲取一些文本,然后我正在做一些轉換,我正在獲取句子和分數,但我的作者過度寫入結果而不是附加,我想知道有沒有為beam.filesystems 添加模塊?from __future__ import absolute_importimport argparseimport loggingfrom datetime import datetimefrom past.builtins import unicodeimport jsonfrom google.cloud import languagefrom google.cloud.language import enumsfrom google.cloud.language import typesimport apache_beam as beamimport apache_beam.transforms.window as windowfrom apache_beam.io.filesystems import FileSystemsfrom apache_beam.io.gcp.pubsub import WriteToPubSubfrom apache_beam.examples.wordcount import WordExtractingDoFnfrom apache_beam.options.pipeline_options import PipelineOptionsfrom apache_beam.options.pipeline_options import SetupOptionsfrom apache_beam.options.pipeline_options import StandardOptionsfrom apache_beam.io.textio import ReadFromText, WriteToTextdef run(argv=None):  """Build and run the pipeline."""  parser = argparse.ArgumentParser()  parser.add_argument(      '--output',         dest='output',        required=True,        help='GCS destination folder to save the images to (example: gs://BUCKET_NAME/path')  group = parser.add_mutually_exclusive_group(required=True)  group.add_argument(      '--input_topic',      help=('Input PubSub topic of the form '            '"projects<project name>/subscriptions/<topic name>".'))  group.add_argument(      '--input_subscription',      help=('Input PubSub subscription of the form '            '"projects<project name>/subscriptions/<subsciption name>."'))  known_args, pipeline_args = parser.parse_known_args(argv)  # We use the save_main_session option because one or more DoFn's in this  # workflow rely on global context (e.g., a module imported at module level).  pipeline_options = PipelineOptions(pipeline_args)  pipeline_options.view_as(SetupOptions).save_main_session = True  pipeline_options.view_as(StandardOptions).streaming = True  p = beam.Pipeline(options=pipeline_options)我只是明白這個:<sentence n> <score>我需要一些小的修復,我被卡住了請幫助我。
查看完整描述

1 回答

?
白板的微信

TA貢獻1883條經驗 獲得超3個贊

為此,您可以嘗試使用beam.io.textio.WriteToText:


messages = (p | "Read From PubSub" >> beam.io.ReadFromPubSub(subscription=known_args.subscription)

    | "Write to GCS" >> beam.io.WriteToText('gs://<your_bucket>/<your_file>', file_name_suffix='.txt',append_trailing_newlines=True,shard_name_template=''))

當您完成流式傳輸作業時,這將為您提供一個文件作為輸出。


希望能幫助到你!


查看完整回答
反對 回復 2021-12-08
  • 1 回答
  • 0 關注
  • 201 瀏覽
慕課專欄
更多

添加回答

舉報

0/150
提交
取消
微信客服

購課補貼
聯系客服咨詢優惠詳情

幫助反饋 APP下載

慕課網APP
您的移動學習伙伴

公眾號

掃描二維碼
關注慕課網微信公眾號