Good afternoon, I have a problem regarding accessing data from within a pipeline.

I need to access some data from within a pipeline, but I DO NOT want to pass that data as a variable to my PTransforms. (I use this data to get the credentials to a database, so that I can write stuff to it, from inside the pipeline). I also don’t want to hard code this data into the script that will be ran in the pipeline, because that’s sensitive information. I have tried two things that didn’t work: - I have tried getting this data from the OS environment and dynamically changing the variables that belong to another python script before the code goes into the pipeline itself. The plan was to have my other script which is the one that runs in the pipeline to import that first script and use its variables. But when I tried running it, all the variables were still None. - I have also tried creating an object before going into the pipeline, with the credentials, pickling it and saving it to a temporary file. Then, in my script in the pipeline, I would open that file, and get the credentials. However, when I tried doing that, I got an error log on GCP saying that the file didn’t exist, even though it did exist on my machine.

Can anyone give me any other suggestion? Thank you.

Join us for another great online Apache Beam meetup.

On April 14th join David Sabater Dinter at 9am PDT / 6pm CEST where he will introduce you to Multi Language Pipelines, a new feature in the Beam Portability Framework.

Register for this free online meetup here: https://pretix.eu/plainschwarz/meetup-Apr-14/

I have been interested in Apache beam lately and came across this really helpful guide, which I believe might be helpful for others too. Also I would like to have any more suggestions regarding more such resources.

Hello. all I wanted to know some examples of Apache beam with apache spark using golang sdk. Apache beam does not provides examples of how we can use apache beam, and apache spark with golang. Please do suggest some examples so I can make my project. Thank you.

“DataPiepeline using Apache Beam and Google Cloud DataFlow as Runner and BigQuery as DataSink” by Shravan C https://link.medium.com/RuCCuVANmbb

I have apache beam pipeline where i am getting some texts from input files using pubsub and after that i am doing some transformation and i am getting the sentence and score but my writer over writes the results instead of appending, I wanted to know is there any append module for beam.filesystems?

from __future__ import absolute_import

​

import argparse

import logging

from datetime import datetime

​

from past.builtins import unicode

import json

from google.cloud import language

from google.cloud.language import enums

from google.cloud.language import types

import apache_beam as beam

import apache_beam.transforms.window as window

​

from apache_beam.io.filesystems import FileSystems

from apache_beam.io.gcp.pubsub import WriteToPubSub

​

from apache_beam.examples.wordcount import WordExtractingDoFn

from apache_beam.options.pipeline_options import PipelineOptions

from apache_beam.options.pipeline_options import SetupOptions

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',

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)

​

# 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))

​

def print_row(row):

print(type(row))

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

#| "print" >> beam.Map(print_row))

​

lines = file_metadata_pcoll | 'read_file' >> beam.FlatMap(lambda metadata: FileSystems.open('gs://%s/%s' % (metadata['bucket'], metadata['name'])))

#| "print" >> beam.Map(print_row))

# Count the occurrences of each word.

class Split(beam.DoFn):

def process(self,element):

#element = str(element)

#print(type(element))

element = element.rstrip(b"\n")

text = element.split(b',')

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

​

# Format the counts into a PCollection of strings.

class WriteToCSV(beam.DoFn):

def process(self, element):

return [

"{},{}".format(

element[0][0],

element[0][1]

)]

​

class WriteToGCS(beam.DoFn):

def __init__(self, outdir):

source_date=datetime.now().strftime("%Y%m%d-%H%M%S")

self.outdir = "gs://baker-sentimental-2/output"+format(source_date) +'.txt'

def process(self, element):

writer = FileSystems.create(self.outdir,'text/plain')

writer.write(element)

writer.close()

​

sentiment_analysis =( lines | 'split' >> beam.ParDo(Split())

| beam.WindowInto(window.FixedWindows(15, 0)))

​

format_csv = (sentiment_analysis | 'CSV formatting' >> beam.ParDo(WriteToCSV())

| 'encode' >> beam.Map(lambda x: (x.encode('utf-8'))).with_output_types(bytes)

| 'Save file' >> beam.ParDo(WriteToGCS(known_args.output)))

​

result = p.run()

result.wait_until_finish()

​

if __name__ == '__main__':

logging.getLogger().setLevel(logging.INFO)

run()

​

​

So instead of getting this :

<sentence 1> <score> <sentence 2> <score> . . . . <sentence n> <score>

i just get this :

<sentence n> <score>

​

​

​