Ufdjrw

I have a GCS bucket where i receive files every minute,So i am using pub/sub method to get new files and doing some transformation and storing into another bucket.but i am getting

"Error received from SDK harness for instruction"

This is my Code :

from __future__ import absolute_import
import os
import logging
import argparse
from google.cloud import language
from google.cloud.language import enums
from google.cloud.language import types
from datetime import datetime
import apache_beam as beam 
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.options.pipeline_options import StandardOptions
from apache_beam.io.textio import ReadFromText, WriteToText

def run(argv=None):
 """Build and run the pipeline."""
 parser = argparse.ArgumentParser()
 parser.add_argument(
 '--output_filename', required=True,
 help=('Output GCS bucket '
 '"gs:/***********/output_files".'))
 group = parser.add_mutually_exclusive_group(required=True)
 group.add_argument(
 '--input_topic',
 help=('Input PubSub topic of the form '
 '"projects/*******/topics/testsub1".'))
 group.add_argument(
 '--input_subscription',
 help=('Input PubSub subscription of the form '
 '"projects/*********/*****/test_subscription."'))
 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(StandardOptions).runner = 'DataflowRunner'
 pipeline_options.view_as(SetupOptions).save_main_session = True
 pipeline_options.view_as(StandardOptions).streaming = True
 p = beam.Pipeline(options=pipeline_options)


 # Read from PubSub into a PCollection.
 if known_args.input_subscription:
 messages = (p
 | beam.io.ReadFromPubSub(
 subscription=known_args.input_subscription)
 .with_output_types(bytes))
 else:
 messages = (p
 | beam.io.ReadFromPubSub(topic=known_args.input_topic)
 .with_output_types(bytes))

 lines = messages | 'decode' >> beam.Map(lambda x: x.decode('utf-8'))

 class Split(beam.DoFn):
 def process(self,element):
 element = element.rstrip("n").encode('utf-8')
 text = element.split(',') 
 result = []
 for i in range(len(text)):
 dat = text[i]
 #print(dat)
 client = language.LanguageServiceClient()
 document = types.Document(content=dat,type=enums.Document.Type.PLAIN_TEXT)
 sent_analysis = client.analyze_sentiment(document=document)
 sentiment = sent_analysis.document_sentiment
 data = [
 (dat,sentiment.score)
 ] 
 result.append(data)
 return result

 class WriteToCSV(beam.DoFn):
 def process(self, element):
 return [
 ",".format(
 element[0][0],
 element[0][1]
 )
 ]

 Transform = (lines
 | 'split' >> beam.ParDo(Split())
 | beam.ParDo(WriteToCSV()) 
 | beam.io.WriteToText(known_args.output_filename)
 )
 result = p.run()
 result.wait_until_finish()

if __name__ == '__main__':
 logging.getLogger().setLevel(logging.INFO)
 run()

I am running this by writing the following code :

python SentAnal.py --runner DataflowRunner --project
b********** --temp_location gs://baker********/tmp/
--input_topic "projects/******/topics/*****" --output_filename "gs://*********/********" --streaming --experiments=allow_non_updatable_job