1

I'm doing a very simple pipeline with dataflow.

It gets a raw data from pubsub and adds a timestamp then write to raw file (I tried parquet first).

Code:

class GetTimestampFn(beam.DoFn):
    """Prints element timestamp"""
    def process(self, element, timestamp=beam.DoFn.TimestampParam):
        timestamp_utc = float(timestamp)
        yield {'raw':str(element),"inserted_at":timestamp_utc}


options = PipelineOptions(streaming=True)
p = beam.Pipeline(DirectRunner(), options=options)
parser = argparse.ArgumentParser()
parser.add_argument('--input_topic',required=True)
parser.add_argument('--output_parquet',required=True)
known_args, _ = parser.parse_known_args(argv)


raw_data = p | 'Read'  >> beam.io.ReadFromPubSub(subscription=known_args.input_topic) 
raw_with_timestamp = raw_data | 'Getting Timestamp' >> beam.ParDo(GetTimestampFn())
_ = raw_with_timestamp | 'Write' >> beam.io.textio.WriteToText(known_args.output_parquet,append_trailing_newlines=True
    ,file_name_suffix='.gzip'
)

p.run().wait_until_finish()

It works with direct runner but it fails on dataflowrunner with this message "Workflow failed."

Job id: 2021-04-14_17_11_02-16453427249129279174

How I'm running the job:

python  real_time_events.py \
--region us-central1 \
--input_topic 'projects/{project}/subscriptions/{subscription}' \
--output_parquet 'gs://{bucket}/stream/' \
--project "{project}" \
--temp_location "gs://{bucket}/tmp" \
--staging_location "gs://{bucket}/stage" 

Any ideas on how to solve ?

2
  • Try by adding --save_main_session in arguments
    – Renaud
    Apr 15, 2021 at 12:39
  • Thanks, just tried that still same error. Apr 15, 2021 at 13:59

1 Answer 1

0

WriteToText does not support streaming pipelines. I recommend you try using fileio.WriteToFiles instead, to get a transform that supports streaming. Note that you may need to group before it.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.