Hi Jan,
it seems that what we would want is to couple the
lifecycle of the Reader not with the restriction but with
the particular instance of (Un)boundedSource (after being
split). That could be done in the processing DoFn, if it
contained a cache mapping instance of the source to the
(possibly null - i.e. not yet open) reader. In
@NewTracker we could assign (or create) the reader to the
tracker, as the tracker is created for each restriction.
WDYT?
I was thinking about this but it seems like it is not
applicable to the way how UnboundedSource and UnboundedReader
work together.
Please correct me if I'm wrong. The UnboundedReader is
created from UnboundedSource per CheckpointMark[1], which
means for certain sources, the CheckpointMark could affect
some attributes like start position of the reader when
resuming. So a single UnboundedSource could be mapped to
multiple readers because of different instances of
CheckpointMarl. That's also the reason why we use
CheckpointMark as the restriction.
Please let me know if I misunderstand your suggestion.
[1]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java#L73-L78
On Mon, Dec 21, 2020 at 9:18 AM Antonio Si
<antonio...@gmail.com <mailto:antonio...@gmail.com>> wrote:
Hi Boyuan,
Sorry for my late reply. I was off for a few days.
I didn't use DirectRunner. I am using FlinkRunner.
We measured the number of Kafka messages that we can
processed per second.
With Beam v2.26 with --experiments=use_deprecated_read
and --fasterCopy=true,
we are able to consume 13K messages per second, but with
Beam v2.26
without the use_deprecated_read option, we are only able
to process 10K messages
per second for the same pipeline.
Thanks and regards,
Antonio.
On 2020/12/11 22:19:40, 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?
> > > >>>>
> > > >>>
> > >
> >
>
