You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Marco Mistroni <mm...@gmail.com> on 2020/04/29 16:18:39 UTC
Beam pipeline parallel steps
Hi all
Is it possible in beam to create a pipeline where two tasks can run in
parallel as opposed to sequential,?
Simple usecase would be step 3 will generate some data out of which I
generate eg 3 completely different outcomes. ( Eg 3 different files stored
in a bucket)
Thanks
Marco
Re: Beam pipeline parallel steps
Posted by Marco Mistroni <mm...@gmail.com>.
Many thanks
On Wed, Apr 29, 2020, 5:56 PM André Rocha Silva <
a.silva@portaltelemedicina.com.br> wrote:
> Hey
>
> You simply use the output PCollection from one to many pipes as you want.
> E.g.:
> p = beam.Pipeline(options=pipeline_options)
>
> data = (
> p
> | 'Get data' >> beam.io.ReadFromText(user_options.input_file)
> )
>
> output1 = (
> data
> | 'Transform 1' >> beam.ParDo(trasnf1())
> | 'Write transform 1 results' >> beam.io.WriteToText(out1)
> )
>
> output2 = (
> data
> | 'Transform 2' >> beam.ParDo(trasnf2())
> | 'Another one' >> beam.FlatMap(something, user_options.parameter)
> | 'Write transform 2 results' >> beam.io.WriteToText(out2)
> )
>
> p.run()
>
>
> On Wed, Apr 29, 2020 at 1:19 PM Marco Mistroni <mm...@gmail.com>
> wrote:
>
>> Hi all
>> Is it possible in beam to create a pipeline where two tasks can run in
>> parallel as opposed to sequential,?
>> Simple usecase would be step 3 will generate some data out of which I
>> generate eg 3 completely different outcomes. ( Eg 3 different files stored
>> in a bucket)
>> Thanks
>> Marco
>>
>
>
> --
>
> *ANDRÉ ROCHA SILVA*
> * DATA ENGINEER*
> (48) 3181-0611
>
> <https://www.linkedin.com/in/andre-rocha-silva/> /andre-rocha-silva/
> <http://portaltelemedicina.com.br/>
> <https://www.youtube.com/channel/UC0KH36-OXHFIKjlRY2GyAtQ>
> <https://pt-br.facebook.com/PortalTelemedicina/>
> <https://www.linkedin.com/company/9426084/>
>
>
Re: Beam pipeline parallel steps
Posted by André Rocha Silva <a....@portaltelemedicina.com.br>.
Hello everybody
I think this will do the job:
p = beam.Pipeline(options=pipeline_options)
data = (
p
| 'Get data' >> beam.io.ReadFromText(user_options.input_file)
)
output1 = (
data
| 'Transform 1' >> beam.ParDo(trasnf1())
)
output2 = (
data
| 'Transform 2' >> beam.ParDo(trasnf2())
| 'Another one' >> beam.FlatMap(something, user_options.parameter)
)
aggregate = (
(output1, output2)
| 'Aggregate both pipes into one' >> beam.Flatten()
| 'Write to file' >> beam.io.WriteToText(out3)
p.run()
Let me know if it worked.
Best
On Sat, May 2, 2020 at 7:04 PM Marco Mistroni <mm...@gmail.com> wrote:
> HI
> thanks and in the example above, what if i want to combine output1 and
> output2 in a unique dta structure that i can then write to teh same file in
> a bucket?
> is there some sort of aggregator in Beam?>
> thanks
>
> On Wed, Apr 29, 2020 at 5:56 PM André Rocha Silva <
> a.silva@portaltelemedicina.com.br> wrote:
>
>> Hey
>>
>> You simply use the output PCollection from one to many pipes as you want.
>> E.g.:
>> p = beam.Pipeline(options=pipeline_options)
>>
>> data = (
>> p
>> | 'Get data' >> beam.io.ReadFromText(user_options.input_file)
>> )
>>
>> output1 = (
>> data
>> | 'Transform 1' >> beam.ParDo(trasnf1())
>> | 'Write transform 1 results' >> beam.io.WriteToText(out1)
>> )
>>
>> output2 = (
>> data
>> | 'Transform 2' >> beam.ParDo(trasnf2())
>> | 'Another one' >> beam.FlatMap(something, user_options.parameter)
>> | 'Write transform 2 results' >> beam.io.WriteToText(out2)
>> )
>>
>> p.run()
>>
>>
>
>> On Wed, Apr 29, 2020 at 1:19 PM Marco Mistroni <mm...@gmail.com>
>> wrote:
>>
>>> Hi all
>>> Is it possible in beam to create a pipeline where two tasks can run in
>>> parallel as opposed to sequential,?
>>> Simple usecase would be step 3 will generate some data out of which I
>>> generate eg 3 completely different outcomes. ( Eg 3 different files stored
>>> in a bucket)
>>> Thanks
>>> Marco
>>>
>>
>>
>> --
>>
>> *ANDRÉ ROCHA SILVA*
>> * DATA ENGINEER*
>> (48) 3181-0611
>>
>> <https://www.linkedin.com/in/andre-rocha-silva/> /andre-rocha-silva/
>> <http://portaltelemedicina.com.br/>
>> <https://www.youtube.com/channel/UC0KH36-OXHFIKjlRY2GyAtQ>
>> <https://pt-br.facebook.com/PortalTelemedicina/>
>> <https://www.linkedin.com/company/9426084/>
>>
>>
--
*ANDRÉ ROCHA SILVA*
* DATA ENGINEER*
(48) 3181-0611
<https://www.linkedin.com/in/andre-rocha-silva/> /andre-rocha-silva/
<http://portaltelemedicina.com.br/>
<https://www.youtube.com/channel/UC0KH36-OXHFIKjlRY2GyAtQ>
<https://pt-br.facebook.com/PortalTelemedicina/>
<https://www.linkedin.com/company/9426084/>
Re: Beam pipeline parallel steps
Posted by Marco Mistroni <mm...@gmail.com>.
HI
thanks and in the example above, what if i want to combine output1 and
output2 in a unique dta structure that i can then write to teh same file in
a bucket?
is there some sort of aggregator in Beam?>
thanks
On Wed, Apr 29, 2020 at 5:56 PM André Rocha Silva <
a.silva@portaltelemedicina.com.br> wrote:
> Hey
>
> You simply use the output PCollection from one to many pipes as you want.
> E.g.:
> p = beam.Pipeline(options=pipeline_options)
>
> data = (
> p
> | 'Get data' >> beam.io.ReadFromText(user_options.input_file)
> )
>
> output1 = (
> data
> | 'Transform 1' >> beam.ParDo(trasnf1())
> | 'Write transform 1 results' >> beam.io.WriteToText(out1)
> )
>
> output2 = (
> data
> | 'Transform 2' >> beam.ParDo(trasnf2())
> | 'Another one' >> beam.FlatMap(something, user_options.parameter)
> | 'Write transform 2 results' >> beam.io.WriteToText(out2)
> )
>
> p.run()
>
>
> On Wed, Apr 29, 2020 at 1:19 PM Marco Mistroni <mm...@gmail.com>
> wrote:
>
>> Hi all
>> Is it possible in beam to create a pipeline where two tasks can run in
>> parallel as opposed to sequential,?
>> Simple usecase would be step 3 will generate some data out of which I
>> generate eg 3 completely different outcomes. ( Eg 3 different files stored
>> in a bucket)
>> Thanks
>> Marco
>>
>
>
> --
>
> *ANDRÉ ROCHA SILVA*
> * DATA ENGINEER*
> (48) 3181-0611
>
> <https://www.linkedin.com/in/andre-rocha-silva/> /andre-rocha-silva/
> <http://portaltelemedicina.com.br/>
> <https://www.youtube.com/channel/UC0KH36-OXHFIKjlRY2GyAtQ>
> <https://pt-br.facebook.com/PortalTelemedicina/>
> <https://www.linkedin.com/company/9426084/>
>
>
Re: Beam pipeline parallel steps
Posted by André Rocha Silva <a....@portaltelemedicina.com.br>.
Hey
You simply use the output PCollection from one to many pipes as you want.
E.g.:
p = beam.Pipeline(options=pipeline_options)
data = (
p
| 'Get data' >> beam.io.ReadFromText(user_options.input_file)
)
output1 = (
data
| 'Transform 1' >> beam.ParDo(trasnf1())
| 'Write transform 1 results' >> beam.io.WriteToText(out1)
)
output2 = (
data
| 'Transform 2' >> beam.ParDo(trasnf2())
| 'Another one' >> beam.FlatMap(something, user_options.parameter)
| 'Write transform 2 results' >> beam.io.WriteToText(out2)
)
p.run()
On Wed, Apr 29, 2020 at 1:19 PM Marco Mistroni <mm...@gmail.com> wrote:
> Hi all
> Is it possible in beam to create a pipeline where two tasks can run in
> parallel as opposed to sequential,?
> Simple usecase would be step 3 will generate some data out of which I
> generate eg 3 completely different outcomes. ( Eg 3 different files stored
> in a bucket)
> Thanks
> Marco
>
--
*ANDRÉ ROCHA SILVA*
* DATA ENGINEER*
(48) 3181-0611
<https://www.linkedin.com/in/andre-rocha-silva/> /andre-rocha-silva/
<http://portaltelemedicina.com.br/>
<https://www.youtube.com/channel/UC0KH36-OXHFIKjlRY2GyAtQ>
<https://pt-br.facebook.com/PortalTelemedicina/>
<https://www.linkedin.com/company/9426084/>
Re: Beam pipeline parallel steps
Posted by Alex Van Boxel <al...@vanboxel.be>.
Yes, this is not a problem. We do it regularly in our streaming pipelines
as a single stream doesn't have enough load for ruining on a Dataflow. So
we run different streams in parallel in a single Beam pipelines.
Data wise the streams have nothing to do with each other, but
transformation wise they are very similar.
On Wed, Apr 29, 2020, 18:19 Marco Mistroni <mm...@gmail.com> wrote:
> Hi all
> Is it possible in beam to create a pipeline where two tasks can run in
> parallel as opposed to sequential,?
> Simple usecase would be step 3 will generate some data out of which I
> generate eg 3 completely different outcomes. ( Eg 3 different files stored
> in a bucket)
> Thanks
> Marco
>