有没有办法在我的python代码中使用apache beam在某个管道之后创建一个空文件答案

【问题标题】：Is there a way to create an empty file after a certain pipeline in my python code using apache beam有没有办法在我的python代码中使用apache beam在某个管道之后创建一个空文件
【发布时间】：2021-02-03 12:43:06
【问题描述】：

下面是我运行管道的python代码。

from __future__ import absolute_import
import apache_beam as beam
import argparse
import logging
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.internal.clients import bigquery
from datetime import date
today = date.today()
current_date = today.strftime("%Y%m%d")
def run(argv=None):
    parser = argparse.ArgumentParser()
    known_args, pipeline_args = parser.parse_known_args(argv)
    p = beam.Pipeline(options=PipelineOptions(pipeline_args))
    (p | 'ReadTable' >> beam.io.Read(beam.io.BigQuerySource(query="SELECT DISTINCT(IF(LENGTH(MOBILE)=10, CONCAT('91',MOBILE),REPLACE(MOBILE,'+91 ','91'))) FROM `whr-asia-datalake-nonprod.WHR_DATALAKE.C4C_CONSUMER_RAW` WHERE REGEXP_CONTAINS(REGEXP_REPLACE(Mobile, ' ', ''),r'^(?:(?:\+|0{0,2})91(\s*[\-]\s*)?|[0]?)?[6789]\d{9}$')",use_standard_sql=True))
       | 'read values' >> beam.Map(lambda x: x.values())
       | 'CSV format' >> beam.Map(lambda row:'|'.join ("WRPOOL|5667788|"+ str(column) +'|"'+"Hi, This msg is from Whirlpool DL" + '"' for column in row))
       | 'Write_to_GCS' >> beam.io.WriteToText('gs://whr-asia-datalake-dev-standard/outbound/Valuefirst/WHR_MOBILE_CNSNT_REQ'+''+ str(current_date),file_name_suffix='.csv',header='SENDER_ID|SHORTCODE|MOBILE_NUM|CONSENT_MSG')
    p.run().wait_until_finish()
if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

在此代码中，创建 csv 文件后，我还需要创建一个空文件。我在我的 csv 文件之后尝试了此选项，但它不会创建一个空文件，而是将 csv 文件名写入其中。

| '创建 .done 文件' >> beam.io.WriteToText('gs://whr-asia-datalake-dev-standard/outbound/Valuefirst/Valuefirst'+str(current_date),file_name_suffix='.done'))

所以我尝试选择 beam.Create('gs://whr-asia-datalake-dev-standard/outbound/Valuefirst/Valuefirst.done')

这是一个错误。任何人都可以帮助创建一个空文件的选项。

【问题讨论】：

标签： python apache-beam dataflow

【解决方案1】：

我认为没有任何内置方法可以创建空文件。您最好的选择是直接使用 Cloud Storage API 在 WriteToText 转换后在 DoFn 中创建空文件

【讨论】：