> Primary purpose of DirectRunner is testing, not
performance
That's one argument, but it's very difficult to
effectively test a pipeline when I need to wait 15+
minutes for the first element to go through it. I
also, disagree in general that we shouldn't care
about the performance of the DirectRunner. It's
likely the first runner new users of beam try (I
know it was for us), and if it doesn't provide
enough performance to actually run a
representative pipeline, users may extrapolate that
performance onto other runners (I know we did).
Anecdotally, the fact that the DirectRunner didn't
work for some of our initial test pipelines (because
of performance problems) probably delayed our
adoption of beam by at least 6 months.
> Steve, based on your findings, it seems like it
takes more time for the SDF pipeline to actually
start to read from PubSub and more time to output
records.
Pubsub reads start ~instantly. but I'm not able to
see any elements actually output from it for a LONG
time, sometimes 30+ minutes. I see the reader
acking back to pubsub, so it IS committing, but no
elements output.
After a bunch of debugging, I think I finally
figured out what the problem is though. During a
checkpoint (in trySplit), the UnboundedSourceViaSDF
wrapper will close the current source reader and
create a new one. The problem is, the pubsub reader
needs some time to correctly estimate it's watermark
[1], and because it gets closed and recreated so
frequently due to checkpointing (either number of
elements, or duration), it can never actually
provide accurate estimates, and always returns the
min watermark. This seems like it prevents some
internal timers from ever firing, effectively
holding all elements in the pipeline state. I can
confirm this also by looking at WatermarkManager,
where I see all the bundles pending.
[1]
https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java#L959
On Thu, Dec 17, 2020 at 9:43 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
Hi Ismaël,
what I meant by the performance vs. testing
argument is that when
choosing default values for certain (possibly
configurable) options, we
should prefer choices that result in better
tested code, not better
performance. DirectRunner actually does quite
many things that are
suboptimal performance-wise, but are good to be
done for test purposes
(immutability checks, as an example).
Regarding SDF in general, I can confirm we see
some issues with Flink,
most recently [1] (which I'm trying to fix right
now). That is actually
correctness, not performance issue. I personally
didn't notice any
performance issues, so far.
Jan
[1] https://issues.apache.org/jira/browse/BEAM-11481
On 12/17/20 3:24 PM, Ismaël Mejía wrote:
> The influence of checkpointing on the output
of the results should be
> minimal in particular for Direct Runner. It
seems what Steve reports
> here seems to be something different. Jan have
you or others already
> checked the influence of this on Flink who is
now using this new
> translation path?
>
> I think the argument that the Direct runner is
mostly about testing
> and not about performance is an argument that
is playing bad on Beam,
> one should not necessarily exclude the other.
Direct runner is our
> most used runner, basically every Beam user
relies on the direct
> runners so every regression or improvement on
it affects everyone, but
> well that's a subject worth its own thread.
>
> On Thu, Dec 17, 2020 at 10:55 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
>> Hi,
>>
>> from my point of view the number in
DirectRunner are set correctly. Primary purpose
of DirectRunner is testing, not performance, so
DirectRunner makes intentionally frequent
checkpoints to easily exercise potential bugs in
user code. It might be possible to make the
frequency configurable, though.
>>
>> Jan
>>
>> On 12/17/20 12:20 AM, Boyuan Zhang wrote:
>>
>> It's not a portable execution on DirectRunner
so I would expect that outputs from
OutputAndTimeBoundedSplittableProcessElementInvoker
should be emitted immediately. For SDF execution
on DirectRunner, the overhead could come from
the SDF expansion, SDF wrapper and the invoker.
>>
>> Steve, based on your findings, it seems like
it takes more time for the SDF pipeline to
actually start to read from PubSub and more time
to output records. Are you able to tell how much
time each part is taking?
>>
>> On Wed, Dec 16, 2020 at 1:53 PM Robert
Bradshaw <[email protected]
<mailto:[email protected]>> wrote:
>>> If all it takes is bumping these numbers up
a bit, that seems like a reasonable thing to do
ASAP. (I would argue that perhaps they shouldn't
be static, e.g. it might be preferable to start
emitting results right away, but use larger
batches for the steady state if there are
performance benefits.)
>>>
>>> That being said, it sounds like there's
something deeper going on here. We should also
verify that this performance impact is limited
to the direct runner.
>>>
>>> On Wed, Dec 16, 2020 at 1:36 PM Steve
Niemitz <[email protected]
<mailto:[email protected]>> wrote:
>>>> I tried changing my build locally to 10
seconds and 10,000 elements but it didn't seem
to make much of a difference, it still takes a
few minutes for elements to begin actually
showing up to downstream stages from the Pubsub
read. I can see elements being emitted from
OutputAndTimeBoundedSplittableProcessElementInvoker,
and bundles being committed by
ParDoEvaluator.finishBundle, but after that,
they seem to just kind of disappear somewhere.
>>>>
>>>> On Wed, Dec 16, 2020 at 4:18 PM Boyuan
Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>> Making it as the PipelineOptions was my
another proposal but it might take some time to
do so. On the other hand, tuning the number into
something acceptable is low-hanging fruit.
>>>>>
>>>>> On Wed, Dec 16, 2020 at 12:48 PM Ismaël
Mejía <[email protected]
<mailto:[email protected]>> wrote:
>>>>>> It sounds reasonable. I am wondering also
on the consequence of these
>>>>>> parameters for other runners (where it is
every 10 seconds or 10000
>>>>>> elements) + their own configuration e.g.
checkpointInterval,
>>>>>> checkpointTimeoutMillis and
minPauseBetweenCheckpoints for Flink. It
>>>>>> is not clear for me what would be chosen
now in this case.
>>>>>>
>>>>>> I know we are a bit anti knobs but maybe
it makes sense to make this
>>>>>> configurable via PipelineOptions at least
for Direct runner.
>>>>>>
>>>>>> On Wed, Dec 16, 2020 at 7:29 PM Boyuan
Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>> I agree, Ismael.
>>>>>>>
>>>>>>> From my current investigation, the
performance overhead should majorly come from
the frequency of checkpoint in
OutputAndTimeBoundedSplittableProcessElementinvoker[1],
which is hardcoded in the DirectRunner(every 1
seconds or 100 elements)[2]. I believe
configuring these numbers on DirectRunner should
improve reported cases so far. My last proposal
was to change the number to every 5 seconds or
10000 elements. What do you think?
>>>>>>>
>>>>>>> [1]
https://github.com/apache/beam/blob/master/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java
>>>>>>> [2]
https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SplittableProcessElementsEvaluatorFactory.java#L178-L181
>>>>>>>
>>>>>>> On Wed, Dec 16, 2020 at 9:02 AM Ismaël
Mejía <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>> I can guess that the same issues
mentioned here probably will affect
>>>>>>>> the usability for people trying Beam's
interactive SQL on Unbounded IO
>>>>>>>> too.
>>>>>>>>
>>>>>>>> We should really take into account that
the performance of the SDF
>>>>>>>> based path should be as good or better
than the previous version
>>>>>>>> before considering its removal
(--experiments=use_deprecated_read) and
>>>>>>>> probably have consensus when this happens.
>>>>>>>>
>>>>>>>>
>>>>>>>> On Fri, Dec 11, 2020 at 11:33 PM Boyuan
Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>> From what I've seen, the direct
runner initiates a checkpoint after every
element output.
>>>>>>>>> That seems like the 1 second limit
kicks in before the output reaches 100 elements.
>>>>>>>>>
>>>>>>>>> I think the original purpose for
DirectRunner to use a small limit on issuing
checkpoint requests is for exercising SDF better
in a small data set. But it brings overhead on a
larger set owing to too many checkpoints. It
would be ideal to make this limit configurable
from pipeline but the easiest approach is that
we figure out a number for most common cases. Do
you think we raise the limit to 1000 elements or
every 5 seconds will help?
>>>>>>>>>
>>>>>>>>> On Fri, Dec 11, 2020 at 2:22 PM Steve
Niemitz <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>> From what I've seen, the direct
runner initiates a checkpoint after every
element output.
>>>>>>>>>>
>>>>>>>>>> On Fri, Dec 11, 2020 at 5:19 PM
Boyuan Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>>> Hi Antonio,
>>>>>>>>>>>
>>>>>>>>>>> Thanks for the details! Which
version of Beam SDK are you using? And are you
using --experiments=beam_fn_api with
DirectRunner to launch your pipeline?
>>>>>>>>>>>
>>>>>>>>>>> For
ReadFromKafkaDoFn.processElement(), it will take
a Kafka topic+partition as input element and a
KafkaConsumer will be assigned to this
topic+partition then poll records continuously.
The Kafka consumer will resume reading and
return from the process fn when
>>>>>>>>>>>
>>>>>>>>>>> There are no available records
currently(this is a feature of SDF which calls
SDF self-initiated checkpoint)
>>>>>>>>>>> The
OutputAndTimeBoundedSplittableProcessElementInvoker
issues checkpoint request to ReadFromKafkaDoFn
for getting partial results. The checkpoint
frequency for DirectRunner is every 100 output
records or every 1 seconds.
>>>>>>>>>>>
>>>>>>>>>>> It seems like either the
self-initiated checkpoint or DirectRunner issued
checkpoint gives you the performance regression
since there is overhead when rescheduling
residuals. In your case, it's more like that the
checkpoint behavior of
OutputAndTimeBoundedSplittableProcessElementInvoker
gives you 200 elements a batch. I want to
understand what kind of performance regression
you are noticing? Is it slower to output the
same amount of records?
>>>>>>>>>>>
>>>>>>>>>>> On Fri, Dec 11, 2020 at 1:31 PM
Antonio Si <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>>>> Hi Boyuan,
>>>>>>>>>>>>
>>>>>>>>>>>> This is Antonio. I reported the
KafkaIO.read() performance issue on the slack
channel a few days ago.
>>>>>>>>>>>>
>>>>>>>>>>>> I am not sure if this is helpful,
but I have been doing some debugging on the SDK
KafkaIO performance issue for our pipeline and I
would like to provide some observations.
>>>>>>>>>>>>
>>>>>>>>>>>> It looks like in my case the
ReadFromKafkaDoFn.processElement() was invoked
within the same thread and every time
kafaconsumer.poll() is called, it returns some
records, from 1 up to 200 records. So, it will
proceed to run the pipeline steps. Each
kafkaconsumer.poll() takes about 0.8ms. So, in
this case, the polling and running of the
pipeline are executed sequentially within a
single thread. So, after processing a batch of
records, it will need to wait for 0.8ms before
it can process the next batch of records again.
>>>>>>>>>>>>
>>>>>>>>>>>> Any suggestions would be appreciated.
>>>>>>>>>>>>
>>>>>>>>>>>> Hope that helps.
>>>>>>>>>>>>
>>>>>>>>>>>> Thanks and regards,
>>>>>>>>>>>>
>>>>>>>>>>>> Antonio.
>>>>>>>>>>>>
>>>>>>>>>>>> On 2020/12/04 19:17:46, Boyuan
Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>>>>> Opened
https://issues.apache.org/jira/browse/BEAM-11403
for tracking.
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Fri, Dec 4, 2020 at 10:52 AM
Boyuan Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> Thanks for the pointer, Steve!
I'll check it out. The execution paths for
>>>>>>>>>>>>>> UnboundedSource and SDF wrapper
are different. It's highly possible that
>>>>>>>>>>>>>> the regression either comes from
the invocation path for SDF wrapper, or
>>>>>>>>>>>>>> the implementation of SDF wrapper
itself.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Fri, Dec 4, 2020 at 6:33 AM
Steve Niemitz <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Coincidentally, someone else in
the ASF slack mentioned [1] yesterday
>>>>>>>>>>>>>>> that they were seeing
significantly reduced performance using KafkaIO.Read
>>>>>>>>>>>>>>> w/ the SDF wrapper vs the
unbounded source. They mentioned they were using
>>>>>>>>>>>>>>> flink 1.9.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
https://the-asf.slack.com/archives/C9H0YNP3P/p1607057900393900
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On Thu, Dec 3, 2020 at 1:56 PM
Boyuan Zhang <[email protected]
<mailto:[email protected]>> wrote:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Hi Steve,
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> I think the major performance
regression comes from
>>>>>>>>>>>>>>>>
OutputAndTimeBoundedSplittableProcessElementInvoker[1],
which will
>>>>>>>>>>>>>>>> checkpoint the DoFn based on
time/output limit and use timers/state to
>>>>>>>>>>>>>>>> reschedule works.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> [1]
>>>>>>>>>>>>>>>>
https://github.com/apache/beam/blob/master/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> On Thu, Dec 3, 2020 at 9:40 AM
Steve Niemitz <[email protected]
<mailto:[email protected]>>
>>>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> I have a pipeline that reads
from pubsub, does some aggregation, and
>>>>>>>>>>>>>>>>> writes to various places.
Previously, in older versions of beam, when
>>>>>>>>>>>>>>>>> running this in the
DirectRunner, messages would go through the pipeline
>>>>>>>>>>>>>>>>> almost instantly, making it
very easy to debug locally, etc.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> However, after upgrading to
beam 2.25, I noticed that it could take on
>>>>>>>>>>>>>>>>> the order of 5-10 minutes for
messages to get from the pubsub read step to
>>>>>>>>>>>>>>>>> the next step in the pipeline
(deserializing them, etc). The subscription
>>>>>>>>>>>>>>>>> being read from has on the
order of 100,000 elements/sec arriving in it.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Setting
--experiments=use_deprecated_read fixes it, and
makes the
>>>>>>>>>>>>>>>>> pipeline behave as it did before.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> It seems like the SDF
implementation in the DirectRunner here is
>>>>>>>>>>>>>>>>> causing some kind of issue,
either buffering a very large amount of data
>>>>>>>>>>>>>>>>> before emitting it in a
bundle, or something else. Has anyone else run
>>>>>>>>>>>>>>>>> into this?
>>>>>>>>>>>>>>>>>
