You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by vamsikrishna korada <ko...@gmail.com> on 2024/08/25 18:30:23 UTC
[Question] Write a PCollection of elements periodically to a file in Flink on Beam application
Hi everyone,
I'm new to Apache Beam and have a question regarding its usage.
I have a scenario where I need to read a stream of elements from a
PCollection and write them to a new file every 5 minutes.
Initially, I considered using Timers and state stores, but I discovered
that Timers are only applicable to KV pairs. If I convert my PCollection
into a key-value pair with a dummy key and then use timers, I encountered
several issues:
1. It introduces an additional shuffle.
2. With all elements sharing the same key, they would be processed by a
single task in the Flink on Beam application. I prefer not to manually
define the number of keys based on load because I plan to run multiple
pipelines, each with varying loads.
One alternative I considered is using a custom executor thread within my
Writer DoFn to flush the records every 5 minutes. However, this approach
would require me to use a lock to make sure only one of the process element
and the flush blocks are running at a time.
Is there a more effective way to accomplish this?
Thanks,
Vamsi
Re: [Question] Write a PCollection of elements periodically to a file in Flink on Beam application
Posted by Wiśniowski Piotr <co...@gmail.com>.
Vamsi,
Actually I do not think it is possible to avoid shuffling just because
you want to put data into 5 minute fixed windows and there is no
ordering on the incoming input data.
I am not sure what is your upstream source or transformation that
creates the pcollection that you want to save, but there is no
assumption that the elements are correlated to time. The only guarantee
is that there will not be any earlier element than watermark.
So in your problem source pcollection can get messages with this
timestamps at same moment into different workers:
-e1, 2024-01-01T00:00:10
-e2, 2024-01-01T00:03:10
-e3, 2024-01-01T00:08:10
-e4, 2024-01-01T00:04:10
-e5, 2024-01-01T00:06:10
-e6, 2024-01-01T00:09:10
Since this is distributed computation there is no possibility to order
the events globally. Also there is no guarantee that all workers will be
processing all time window buckets.
What can be done is that the worker that receives the message checks if
it is responsible for computing this message window and stores it on a
disc or forwards the message to the worker that is actually responsible
for the message window (which is really a shuffle operation).
What could be done to optimize more the code with sharding prefix is try
to use `GroupIntoBatches.WithShardedKey`(impl ref
https://github.com/apache/beam/blob/bcee5d081d05841bb52f0851f94a04f9c8968b88/sdks/python/apache_beam/transforms/util.py#L1113)
as some runners might optimize this shuffle so that minimum amount of
data is actually sent thru network.
You may try to use that transform (if it will meed your requirements) or
take a look there how `ShardedKey` is used there.
On the idea of custom executor thread - this is asking for troubles.
Typically beam (and all other distributed computation engines) are
designed to handle parallelization for you and if you try to actually do
something with threads on worker things typically end badly (random
failures, missing data, hard debugging). I really doubt it would be more
performant than actual shuffle with beam way.
Best,
Wisniowski Piotr
On 27.08.2024 20:35, vamsikrishna korada wrote:
> Thanks for the reply Wisniowski Piotr.
>
> As you mentioned, the data can be big and I am okay with writing
> multiple files every 5 minutes..
> If I assign a key that uniquely represents a worker, that makes sure
> all the data that is processed together stays together. But there can
> still be a shuffle and the data can collectively be moved across nodes
> when we are assigning the keys right? Is there any way I can avoid the
> extra shuffle?
>
> Do you think it is a good idea to instead have a separate scheduler
> thread within the DoFn which will periodically flush the data every 5
> minutes from each worker?
> Thanks,
> Vamsi
>
> On Tue, 27 Aug 2024 at 17:49, Wiśniowski Piotr
> <co...@gmail.com> wrote:
>
> Hi,
>
> If you require only a single file per 5 min then what you are
> missing is that you need to window data into fixed windows, so
> that state-full DoFn could store the elements per key per window.
>
> Something like (did not test this code just pseudo code):
>
> classStatefulWriteToFileDoFn(beam.DoFn):
> BAG_STATE = BagStateSpec('bag_state', VarIntCoder())
> TIMER = TimerSpec('timer', TimeDomain.WATERMARK)
> defprocess(self, element, timestamp=beam.DoFn.TimestampParam,
> window=beam.DoFn.WindowParam,
> bag_state=beam.DoFn.StateParam(BAG_STATE),
> timer=beam.DoFn.TimerParam(TIMER)):
> bag_state.add(element)
> timer.set(window.end)
> defon_timer(self, window=beam.DoFn.WindowParam,
> bag_state=beam.DoFn.StateParam(BAG_STATE)):
> # Here, you can generate a filename based on the window's end
> time, for example
> filename =
> f'output_{window.end.to_utc_datetime().strftime("%Y%m%d%H%M%S")}.txt'
> withopen(filename, 'w') asf:
> forelement inbag_state.read():
> f.write(f'{element}\n')
> defrun():
> withbeam.Pipeline(options=PipelineOptions()) asp:
> (p
> |'ReadStream'>>beam.io.ReadFromPubSub(topic='projects/.../topics/...')
> |'WindowIntoFixed'>>beam.WindowInto(FixedWindows(300)) # 5-minute
> windows
> |'MapToSingleKey'>>beam.Map(lambdax: (1, x)) # Map to a single key
> |'StatefulWriteToFile'>>beam.ParDo(StatefulWriteToFileDoFn())
> )
>
> So the state-full dofn will keep its state per key per window. So
> each window should crate its own output file. But note that this
> requires you to put all the 5 min data in same worker as only a
> single worker can be responsible for creating the file and hence
> the shuffling is required.
>
> There is a workaround If the traffic might be too big, but it
> would mean to generate more files per 5 min window (one file per
> worker).
>
> The trick is to assign key that uniquely represents the worker,
> not the data. So every worker that maps the key should have his
> unique value put in this key.
> See this example:
>
> class_Key(DoFn):
> defsetup(self):
> self.shard_prefix=str(uuid4())
> defprocess(
> self,
> x: input_type,
> ) -> Iterable[tuple[str, input_type]]:
> yield(
> self.shard_prefix+str(threading.get_ident()), # each worker may
> create his batch
> x,
> )
> And then you can use it like
> |"KeyPerWorker">>ParDo(_Key())
> instead of using constant key with the first approach. Also
> remember to make sure file names are unique if using this approach.
> Best
> Wisniowski Piotr
>
> On 25.08.2024 20:30, vamsikrishna korada wrote:
>>
>> Hi everyone,
>>
>> I'm new to Apache Beam and have a question regarding its usage.
>>
>> I have a scenario where I need to read a stream of elements from
>> a PCollection and write them to a new file every 5 minutes.
>>
>> Initially, I considered using Timers and state stores, but I
>> discovered that Timers are only applicable to KV pairs. If I
>> convert my PCollection into a key-value pair with a dummy key and
>> then use timers, I encountered several issues:
>>
>> 1. It introduces an additional shuffle.
>> 2. With all elements sharing the same key, they would be
>> processed by a single task in the Flink on Beam application.
>> I prefer not to manually define the number of keys based on
>> load because I plan to run multiple pipelines, each with
>> varying loads.
>>
>> One alternative I considered is using a custom executor thread
>> within my Writer DoFn to flush the records every 5 minutes.
>> However, this approach would require me to use a lock to make
>> sure only one of the process element and the flush blocks are
>> running at a time.
>>
>> Is there a more effective way to accomplish this?
>>
>>
>>
>> Thanks,
>>
>> Vamsi
>>
>>
>>
Re: [Question] Write a PCollection of elements periodically to a file in Flink on Beam application
Posted by vamsikrishna korada <ko...@gmail.com>.
Thanks for the reply Wisniowski Piotr.
As you mentioned, the data can be big and I am okay with writing multiple
files every 5 minutes..
If I assign a key that uniquely represents a worker, that makes sure all
the data that is processed together stays together. But there can still be
a shuffle and the data can collectively be moved across nodes when we are
assigning the keys right? Is there any way I can avoid the extra shuffle?
Do you think it is a good idea to instead have a separate scheduler thread
within the DoFn which will periodically flush the data every 5 minutes from
each worker?
Thanks,
Vamsi
On Tue, 27 Aug 2024 at 17:49, Wiśniowski Piotr <
contact.wisniowskipiotr@gmail.com> wrote:
> Hi,
>
> If you require only a single file per 5 min then what you are missing is
> that you need to window data into fixed windows, so that state-full DoFn
> could store the elements per key per window.
>
> Something like (did not test this code just pseudo code):
> class StatefulWriteToFileDoFn(beam.DoFn):
> BAG_STATE = BagStateSpec('bag_state', VarIntCoder())
> TIMER = TimerSpec('timer', TimeDomain.WATERMARK)
> def process(self, element, timestamp=beam.DoFn.TimestampParam,
> window=beam.DoFn.WindowParam,
> bag_state=beam.DoFn.StateParam(BAG_STATE),
> timer=beam.DoFn.TimerParam(TIMER)):
> bag_state.add(element)
> timer.set(window.end)
> def on_timer(self, window=beam.DoFn.WindowParam, bag_state=
> beam.DoFn.StateParam(BAG_STATE)):
> # Here, you can generate a filename based on the window's end time, for
> example
> filename = f'output_{window.end.to_utc_datetime().strftime("%Y%m%d%H%M%S")
> }.txt'
> with open(filename, 'w') as f:
> for element in bag_state.read():
> f.write(f'{element}\n')
> def run():
> with beam.Pipeline(options=PipelineOptions()) as p:
> (p
> | 'ReadStream' >> beam.io.ReadFromPubSub(topic='projects/.../topics/...')
> | 'WindowIntoFixed' >> beam.WindowInto(FixedWindows(300)) # 5-minute
> windows
> | 'MapToSingleKey' >> beam.Map(lambda x: (1, x)) # Map to a single key
> | 'StatefulWriteToFile' >> beam.ParDo(StatefulWriteToFileDoFn())
> )
>
> So the state-full dofn will keep its state per key per window. So each
> window should crate its own output file. But note that this requires you to
> put all the 5 min data in same worker as only a single worker can be
> responsible for creating the file and hence the shuffling is required.
>
> There is a workaround If the traffic might be too big, but it would mean
> to generate more files per 5 min window (one file per worker).
>
> The trick is to assign key that uniquely represents the worker, not the
> data. So every worker that maps the key should have his unique value put in
> this key.
> See this example:
> class _Key(DoFn):
> def setup(self):
> self.shard_prefix = str(uuid4())
> def process(
> self,
> x: input_type,
> ) -> Iterable[tuple[str, input_type]]:
> yield (
> self.shard_prefix + str(threading.get_ident()), # each worker may create
> his batch
> x,
> )
> And then you can use it like
> | "KeyPerWorker" >> ParDo(_Key())
> instead of using constant key with the first approach. Also remember to
> make sure file names are unique if using this approach.
> Best
> Wisniowski Piotr
>
> On 25.08.2024 20:30, vamsikrishna korada wrote:
>
> Hi everyone,
>
> I'm new to Apache Beam and have a question regarding its usage.
>
> I have a scenario where I need to read a stream of elements from a
> PCollection and write them to a new file every 5 minutes.
>
> Initially, I considered using Timers and state stores, but I discovered
> that Timers are only applicable to KV pairs. If I convert my PCollection
> into a key-value pair with a dummy key and then use timers, I encountered
> several issues:
>
> 1. It introduces an additional shuffle.
> 2. With all elements sharing the same key, they would be processed by
> a single task in the Flink on Beam application. I prefer not to manually
> define the number of keys based on load because I plan to run multiple
> pipelines, each with varying loads.
>
> One alternative I considered is using a custom executor thread within my
> Writer DoFn to flush the records every 5 minutes. However, this approach
> would require me to use a lock to make sure only one of the process element
> and the flush blocks are running at a time.
>
> Is there a more effective way to accomplish this?
>
>
>
> Thanks,
>
> Vamsi
>
>
>
>
Re: [Question] Write a PCollection of elements periodically to a file in Flink on Beam application
Posted by Wiśniowski Piotr <co...@gmail.com>.
Hi,
If you require only a single file per 5 min then what you are missing is
that you need to window data into fixed windows, so that state-full DoFn
could store the elements per key per window.
Something like (did not test this code just pseudo code):
classStatefulWriteToFileDoFn(beam.DoFn):
BAG_STATE = BagStateSpec('bag_state', VarIntCoder())
TIMER = TimerSpec('timer', TimeDomain.WATERMARK)
defprocess(self, element, timestamp=beam.DoFn.TimestampParam,
window=beam.DoFn.WindowParam,
bag_state=beam.DoFn.StateParam(BAG_STATE),
timer=beam.DoFn.TimerParam(TIMER)):
bag_state.add(element)
timer.set(window.end)
defon_timer(self, window=beam.DoFn.WindowParam,
bag_state=beam.DoFn.StateParam(BAG_STATE)):
# Here, you can generate a filename based on the window's end time, for
example
filename =
f'output_{window.end.to_utc_datetime().strftime("%Y%m%d%H%M%S")}.txt'
withopen(filename, 'w') asf:
forelement inbag_state.read():
f.write(f'{element}\n')
defrun():
withbeam.Pipeline(options=PipelineOptions()) asp:
(p
|'ReadStream'>>beam.io.ReadFromPubSub(topic='projects/.../topics/...')
|'WindowIntoFixed'>>beam.WindowInto(FixedWindows(300)) # 5-minute windows
|'MapToSingleKey'>>beam.Map(lambdax: (1, x)) # Map to a single key
|'StatefulWriteToFile'>>beam.ParDo(StatefulWriteToFileDoFn())
)
So the state-full dofn will keep its state per key per window. So each
window should crate its own output file. But note that this requires you
to put all the 5 min data in same worker as only a single worker can be
responsible for creating the file and hence the shuffling is required.
There is a workaround If the traffic might be too big, but it would mean
to generate more files per 5 min window (one file per worker).
The trick is to assign key that uniquely represents the worker, not the
data. So every worker that maps the key should have his unique value put
in this key.
See this example:
class_Key(DoFn):
defsetup(self):
self.shard_prefix=str(uuid4())
defprocess(
self,
x: input_type,
) -> Iterable[tuple[str, input_type]]:
yield(
self.shard_prefix+str(threading.get_ident()), # each worker may create
his batch
x,
)
And then you can use it like
|"KeyPerWorker">>ParDo(_Key())
instead of using constant key with the first approach. Also remember to
make sure file names are unique if using this approach.
Best
Wisniowski Piotr
On 25.08.2024 20:30, vamsikrishna korada wrote:
>
> Hi everyone,
>
> I'm new to Apache Beam and have a question regarding its usage.
>
> I have a scenario where I need to read a stream of elements from a
> PCollection and write them to a new file every 5 minutes.
>
> Initially, I considered using Timers and state stores, but I
> discovered that Timers are only applicable to KV pairs. If I convert
> my PCollection into a key-value pair with a dummy key and then use
> timers, I encountered several issues:
>
> 1. It introduces an additional shuffle.
> 2. With all elements sharing the same key, they would be processed by
> a single task in the Flink on Beam application. I prefer not to
> manually define the number of keys based on load because I plan to
> run multiple pipelines, each with varying loads.
>
> One alternative I considered is using a custom executor thread within
> my Writer DoFn to flush the records every 5 minutes. However, this
> approach would require me to use a lock to make sure only one of the
> process element and the flush blocks are running at a time.
>
> Is there a more effective way to accomplish this?
>
>
>
> Thanks,
>
> Vamsi
>
>
>