flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Ufuk Celebi <...@apache.org>
Subject Re: [DISCUSS] FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime
Date Sun, 02 Jul 2017 20:13:53 GMT
Thanks for the write up and illustrations. :-) +1 to do this.

I'm fine with both proposed "changed behaviour" variants, but lean
towards option 1: change the default, make the change explicit in the
release notes and add a good docs page about configuring object reuse
(ideally re-using your illustrations from the FLIP).

I see that option 2 (keep COPY_PER_OPERATOR as default for upgraded
jobs if nothing else is configured) is nice in order to prevent any
surprises for users upgrading from 1.3 to 1.4. But if I understand it
correctly we only postpone the problem to their first 1.4 savepoint +
restore at which point the behaviour would still change, right? If the
answer is yes, I think that this might be more confusing than simply
changing the default (option 1) in the long run.

– Ufuk


On Sun, Jul 2, 2017 at 6:12 PM, Stephan Ewen <sewen@apache.org> wrote:
> Thank you for the reply and for the support!
>
> @Greg, controlling object reuse on a per-operator base is definitely a good
> way to follow up. My first thought would be to keep this proposal slim and
> deal with the "default" logic, and have a followup effort to make this
> controllable per operator.
>
> @Greg When you mention the "surprises" about object reuse in the DataSet
> API, what cases and behavior do you have in mind there?
>
> Stephan
>
>
> On Wed, Jun 28, 2017 at 2:56 PM, Aljoscha Krettek <aljoscha@apache.org>
> wrote:
>
>> +1 for changing the default if so many people encountered problems with
>> serialisation costs.
>>
>> The first two modes don’t require any code changes, correct? Only the last
>> one would require changes to the stream input processors.
>>
>> We should also keep this issue in mind: https://issues.apache.org/
>> jira/browse/FLINK-3974 <https://issues.apache.org/jira/browse/FLINK-3974>
>> i.e. we always need to make shallow copies of the StreamRecord.
>>
>> Best,
>> Aljoscha
>>
>> > On 27. Jun 2017, at 21:01, Zhenzhong Xu <flyfree@gmail.com> wrote:
>> >
>> > Stephan,
>> >
>> > Fully supporting this FLIP. We originally encountered pretty big
>> surprises observing the object copy behavior causing significant
>> performance degradation for our massively parallel use case.
>> >
>> > In our case, (I think most appropriately SHOULD be the assumptions for
>> all streaming use case), is to assume object immutability for all the
>> records throughout processing pipeline, as a result, I don't see a need to
>> distinguish different object reuse behaviors for different (chained)
>> operators (or to the very extreme even the need to support
>> COPY_PER_OPERATOR other than we probably have to support for backward
>> compatibility). I am also a fan of failing fast if user asserts incorrect
>> assumptions.
>> >
>> > One feedback on the FLIP-21 itself, I am not very clear on the
>> difference between DEFAULT and FULL_REUSE enumeration, aren't them exactly
>> the same thing in new proposal? However, the model figures seem to indicate
>> they are slightly different? Can you elaborate a bit more?
>> >
>> > Z.
>> >
>> >
>> > On 2017-06-27 11:14 (-0700), Greg Hogan <code@greghogan.com <mailto:
>> code@greghogan.com>> wrote:
>> >> Hi Stephan,
>> >>
>> >> Would this be an appropriate time to discuss allowing reuse to be a
>> per-operator configuration? Object reuse for chained operators has lead to
>> considerable surprise for some users of the DataSet API. This came up
>> during the rework of the object reuse documentation for the DataSet API.
>> With annotations a Function could mark whether input/iterator or
>> output/collected objects should be copied or reused.
>> >>
>> >> My distant observation is that is is safer to locally assert reuse at
>> the operator level than to assume or guarantee the safety of object reuse
>> across an entire program. It could also be handy to mix operators receiving
>> copyable objects with operators not requiring copyable objects.
>> >>
>> >> Greg
>> >>
>> >>
>> >>> On Jun 27, 2017, at 1:21 PM, Stephan Ewen <sewen@apache.org> wrote:
>> >>>
>> >>> Hi all!
>> >>>
>> >>> I would like to propose the following FLIP:
>> >>>
>> >>> FLIP-21 - Improve object Copying/Reuse Mode for Streaming Runtime:
>> >>> https://cwiki.apache.org/confluence/pages/viewpage.
>> action?pageId=71012982
>> >>>
>> >>> The FLIP is motivated by the fact that many users run into an
>> unnecessary
>> >>> kind of performance problem caused by an old design artifact.
>> >>>
>> >>> The required change should be reasonably small, and would help many
>> users
>> >>> and Flink's general standing.
>> >>>
>> >>> Happy to hear thoughts!
>> >>>
>> >>> Stephan
>> >>>
>> >>> ======================================
>> >>>
>> >>> FLIP text is below. Pictures with illustrations are only in the Wiki,
>> not
>> >>> supported on the mailing list.
>> >>> ------------------------------------------------------------
>> -------------------------------------
>> >>>
>> >>> Motivation
>> >>>
>> >>> The default behavior of the streaming runtime is to copy every element
>> >>> between chained operators.
>> >>>
>> >>> That operation was introduced for “safety† reasons, to avoid
the
>> number of
>> >>> cases where users can create incorrect programs by reusing mutable
>> objects
>> >>> (a discouraged pattern, but possible). For example when using state
>> >>> backends that keep the state as objects on heap, reusing mutable
>> objects
>> >>> can theoretically create cases where the same object is used in
>> multiple
>> >>> state mappings.
>> >>>
>> >>> The effect is that many people that try Flink get much lower
>> performance
>> >>> than they could possibly get. From empirical evidence, almost all users
>> >>> that I (Stephan) have been in touch with eventually run into this issue
>> >>> eventually.
>> >>>
>> >>> There are multiple observations about that design:
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  Object copies are extremely costly. While some simple copy virtually
>> for
>> >>>  free (types reliably detected as immutable are not copied at all),
>> many
>> >>>  real pipelines use types like Avro, Thrift, JSON, etc, which are very
>> >>>  expensive to copy.
>> >>>
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  Keyed operations currently only occur after shuffles. The operations
>> are
>> >>>  hence the first in a pipeline and will never have a reused object
>> anyways.
>> >>>  That means for the most critical operation, this pre-caution is
>> unnecessary.
>> >>>
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  The mode is inconsistent with the contract of the DataSet API, which
>> >>>  does not copy at each step
>> >>>
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  To prevent these copies, users can select {{enableObjectReuse()}},
>> which
>> >>>  is misleading, since it does not really reuse mutable objects, but
>> avoids
>> >>>  additional copies.
>> >>>
>> >>>
>> >>> Proposal
>> >>>
>> >>> Summary
>> >>>
>> >>> I propose to change the default behavior of the DataStream runtime to
>> be
>> >>> the same as the DataSet runtime. That means that new objects are
>> chosen on
>> >>> every deserialization, and no copies are made as the objects are
>> passed on
>> >>> along the pipelines.
>> >>>
>> >>> Details
>> >>>
>> >>> I propose to drop the execution config flag {{objectReuse}} and instead
>> >>> introduce an {{ObjectReuseMode}} enumeration with better control of
>> what
>> >>> should happen. There will be three different types:
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  DEFAULT
>> >>>  -
>> >>>
>> >>>     This is the default in the DataSet API
>> >>>     -
>> >>>
>> >>>     This will become the default in the DataStream API
>> >>>     -
>> >>>
>> >>>     This happens in the DataStream API when {{enableObjectReuse()}}
is
>> >>>     activated.
>> >>>
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  COPY_PER_OPERATOR
>> >>>  -
>> >>>
>> >>>     The current default in the DataStream API
>> >>>
>> >>>
>> >>>
>> >>>  -
>> >>>
>> >>>  FULL_REUSE
>> >>>  -
>> >>>
>> >>>     This happens in the DataSet API when {{enableObjectReuse()}} is
>> >>>     chosen.
>> >>>
>> >>>
>> >>> An illustration of the modes is as follows:
>> >>>
>> >>> DEFAULT
>> >>>
>> >>>
>> >>> See here:
>> >>> https://cwiki.apache.org/confluence/pages/viewpage.
>> action?pageId=71012982&preview=/https%3A%2F%2Flh5.googleusercontent.com%
>> 2F1UOpVB2wSMhx8067IE9t2_mJG549IoOkDiAfIN_uXQZVUvAXCp-hQLY-mgoSWunwF-
>> xciZuJ4pZpj1FX0ZPQrd-Fm1jWzgX3Hv7-SELUdPUvEN6XUPbLrwfA9YRl605bFKMYlf1r <
>> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&
>> preview=/https%3A%2F%2Flh5.googleusercontent.com%2F1UOpVB2wSMhx8067IE9t2_
>> mJG549IoOkDiAfIN_uXQZVUvAXCp-hQLY-mgoSWunwF-xciZuJ4pZpj1FX0ZPQrd-
>> Fm1jWzgX3Hv7-SELUdPUvEN6XUPbLrwfA9YRl605bFKMYlf1r>
>> >>>
>> >>> COPY_PER_OPERATOR
>> >>>
>> >>>
>> >>> See here:
>> >>> https://cwiki.apache.org/confluence/pages/viewpage.
>> action?pageId=71012982&preview=/https%3A%2F%2Flh3.googleusercontent.com%
>> 2Fs5sBOktzaKrRw3v1-IQMgImYZfchQMVz2HiG3i050xCWNTK
>> uQV6mmlv3QtR0TZ0SGPRSCyjI-sUAqfbJw4fGOxKqBuRX2f-
>> iZGh0e7hBke7DzuApUNy1vaF2SgtQVH3XEXkRx8Ks <https://cwiki.apache.org/
>> confluence/pages/viewpage.action?pageId=71012982&
>> preview=/https%3A%2F%2Flh3.googleusercontent.com%2Fs5sBOktzaKrRw3v1-
>> IQMgImYZfchQMVz2HiG3i050xCWNTKuQV6mmlv3QtR0TZ0SGPRSCyjI-
>> sUAqfbJw4fGOxKqBuRX2f-iZGh0e7hBke7DzuApUNy1vaF2SgtQVH3XEXkRx8Ks>
>> >>>
>> >>>
>> >>> FULL_REUSE
>> >>>
>> >>>
>> >>> See here:
>> >>> https://cwiki.apache.org/confluence/pages/viewpage.
>> action?pageId=71012982&preview=/https%3A%2F%2Flh5.googleusercontent.com%
>> 2FFdOzuuaioooEIOh7bo0gZ5dHZrlgEKiwtNjGE9DjR-fT20B0q7FGDAvAk5oh1h58WtNQktuF
>> GinrV1q1Yq8H8ayCyyqFUq-gmAYYW91x4XZQNrjLc6eJ0cptzvN_r8cU_GVV7LNE <
>> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71012982&
>> preview=/https%3A%2F%2Flh5.googleusercontent.com%
>> 2FFdOzuuaioooEIOh7bo0gZ5dHZrlgEKiwtNjGE9DjR-fT20B0q7FGDAvAk5oh1h58WtNQktuF
>> GinrV1q1Yq8H8ayCyyqFUq-gmAYYW91x4XZQNrjLc6eJ0cptzvN_r8cU_GVV7LNE>
>> >>> New or Changed Public Interfaces
>> >>>
>> >>> Interfaces changed
>> >>>
>> >>> The interface of the {{ExecutionConfig}} add the method
>> >>> {{setObjectReuseMode(ObjectReuseMode)}}, and deprecates the methods
>> >>> {{enableObjectReuse()}} and {{disableObjectReuse()}}.
>> >>>
>> >>>
>> >>> Behavior changed
>> >>>
>> >>> The default object passing behavior changes, meaning that it can
>> affect the
>> >>> correctness of prior DataStream programs that assume the original
>> >>> “COPY_PER_OPERATOR† behavior.
>> >>>
>> >>> Migration Plan and Compatibility
>> >>>
>> >>> Interfaces
>> >>>
>> >>> No interface migration path is needed, because the interfaces are not
>> >>> broken, merely some methods get deprecated.
>> >>>
>> >>> Behavior change
>> >>>
>> >>> Variant 1:
>> >>>
>> >>>  -
>> >>>
>> >>>  Change the behavior, make it explicit on the release notes that we
did
>> >>>  that and what cases are affected.
>> >>>  -
>> >>>
>> >>>  This may actually be feasible, because the cases that are affected
are
>> >>>  quite pathological corner cases that only very bad implementations
>> should
>> >>>  encounter (see below)
>> >>>
>> >>>
>> >>> Variant 2:
>> >>>
>> >>>  -
>> >>>
>> >>>  When users set the mode, always that mode is used.
>> >>>  -
>> >>>
>> >>>  When the mode is not explicitly set, we follow that strategy:
>> >>>  -
>> >>>
>> >>>     Change the CLI such that we know when users upgrade existing jobs
>> >>>     (the savepoint to start from has a version prior to 1.4).
>> >>>     -
>> >>>
>> >>>     Use DEFAULT as the default for jobs that do not start from
>> savepoint,
>> >>>     or that start from savepoint >= 1.4
>> >>>     -
>> >>>
>> >>>     Use COPY_PER_OPERATOR as the default for upgraded jobs
>>
>>

Mime
View raw message