> 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ý
<je...@seznam.cz <mailto:je...@seznam.cz>> 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ý <je...@seznam.cz
<mailto:je...@seznam.cz>> 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 <rober...@google.com
<mailto:rober...@google.com>> 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 <sniem...@apache.org
<mailto:sniem...@apache.org>> 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 <boyu...@google.com
<mailto:boyu...@google.com>> 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 <ieme...@gmail.com
<mailto:ieme...@gmail.com>> 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 <boyu...@google.com
<mailto:boyu...@google.com>> 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 <ieme...@gmail.com
<mailto:ieme...@gmail.com>> 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 <boyu...@google.com
<mailto:boyu...@google.com>> 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 <sniem...@apache.org
<mailto:sniem...@apache.org>> 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 <boyu...@google.com
<mailto:boyu...@google.com>> 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 <antonio...@gmail.com
<mailto:antonio...@gmail.com>> 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 <boyu...@google.com
<mailto:boyu...@google.com>> wrote:
>>>>>>>>>>>>> Opened
https://issues.apache.org/jira/browse/BEAM-11403
for tracking.
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Fri, Dec 4, 2020 at 10:52
AM Boyuan Zhang <boyu...@google.com
<mailto:boyu...@google.com>> 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 <sniem...@apache.org
<mailto:sniem...@apache.org>> 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 <boyu...@google.com
<mailto:boyu...@google.com>> 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 <sniem...@apache.org
<mailto:sniem...@apache.org>>
>>>>>>>>>>>>>>>> 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?
>>>>>>>>>>>>>>>>>
