you made a really good argument ;) I'm inclined to an experimental
opt-in flag that would enable this. It would be great if we could
automatically check for violations - kind of a safety net, for
mistakes in user code.
Just to note, direct runner enforcement may not cover all cases, as
it only checks binary representation after serialization. Also there
are programmers that don't write tests, especially during
prototyping (not an argument for perf. penalty, but something to
keep in mind).
Max, WDYT?
On Tue, Oct 27, 2020 at 12:44 PM Teodor Spæren
<teodor_spae...@riseup.net <mailto:teodor_spae...@riseup.net>>
wrote:
Some more thoughts:
As it says on the DirectRunner [1] page, the DirectRunner is meant to
check that users don't rely on semantics that are not guaranteed
by the
Beam model.
Programs that rely on the Flink runner deep cloning the inputs
between
each operator in the pipeline is relying on a semantic that is not
guaranteed by the Beam model, and those pipelines would fail if
ran on
the DirectRunner.
As I stated in the previous email, I have some example programs that
return different outputs on the Flink runner and on the
DirectRunner. I
have not tested these programs on other runners, so I don't know what
they would return. If they return different answers than the
DirectRunner, I'm inclined to say that the DirectRunner should
either be
changed, or the runners be changed.
From my very limited point of view, the Flink runner seems to be
spending a lot of extra time implementing a semantic guarantee
that the
Beam model explicitly doesn't support.
Best regards,
Teodor Spæren
[1]: https://beam.apache.org/documentation/runners/direct/
On Tue, Oct 27, 2020 at 12:08:51PM +0100, Teodor Spæren wrote:
>Hey David,
>
>I think I might have worded this poorly, because what I meant is
that
>from what I can see in [1], the BEAM model explicitly states that
>PCollections should be treated as immutable. The direct runner also
>tests for this. Do the other runners also protect the user from
>misusing the system so? If not we have a situation where running the
>same pipeline on two different runners will yield different
answers. I
>can show some examples that return different examples for the Flink
>and the Direct Runner.
>
>I agree that a breaking existing pipelines is a no-no, but I do
think
>that we could simply gate this behind an option on the Flink runner.
>
>I also tried to search for this before, but did not find any mention
>of it, can you link me to some discussions about this in the past?
>
>Thanks for reply :D
>
>Best regards,
>Teodor Spæren
>
>[1]:
https://beam.apache.org/documentation/programming-guide/#immutability
>
>
>On Tue, Oct 27, 2020 at 11:49:45AM +0100, David Morávek wrote:
>>Hi Teodor,
>>
>>Thanks for bringing this up. This is a known, long standing "issue".
>>Unfortunately there are few things we need to consider:
>>
>>- As you correctly noted, the *Beam model doesn't enforce
immutability* of
>>input / output elements, so this is the price.
>>- We* can not break *existing pipelines.
>>- Flink Runner needs to provide the *same guarantees as the Beam
model*.
>>
>>There are definitely some things we can do here, to make things
faster:
>>
>>- We can try the similar approach as HadoopIO
>>(HadoopInputFormatReader#isKnownImmutable), to check for known
immutable
>>types (KV, primitives, protobuf, other known internal immutable
structures).
>>-* If the type is immutable, we can safely reuse it.* This
should cover
>>most of the performance costs without breaking the guarantees
Beam model
>>provides.
>>- We can enable registration of custom "immutable" types via
pipeline
>>options? (this may be an unnecessary knob, so this needs a further
>>discussion)
>>
>>WDYT?
>>
>>D.
>>
>>
>>On Mon, Oct 26, 2020 at 6:37 PM Teodor Spæren
<teodor_spae...@riseup.net <mailto:teodor_spae...@riseup.net>>
>>wrote:
>>
>>>Hey!
>>>
>>>I'm a student at the University of Oslo, and I'm writing a
master thesis
>>>about the possibility of using Beam to benchmark stream processing
>>>systems. An important factor in this is the overhead associated
with
>>>using Beam over writing code for the runner directly. [1] found
that
>>>there was a large overhead associated with using Beam, but did not
>>>investigate where this overhead came from. I've done benchmarks and
>>>confirmed the findings there, where for simple chains of identity
>>>operators, Beam is 43x times slower than the Flink equivalent.
>>>
>>>These are very simple pipelines, with custom sources that just
output a
>>>series of integers. By profiling I've found that most of the
overhead
>>>comes from serializing and deserializing. Specifically the way
>>>TypeSerializer's, [2], is implemented in [3], where each object is
>>>serialized and then deserialized between every operator.
Looking into
>>>the semantics of Beam, no operator should change the input, so
we don't
>>>need to do a copy here. The function in [3] could potentially
be changed
>>>to a single `return` statement.
>>>
>>>Doing this removes 80% of the overhead in my tests. This is a very
>>>synthetic example, but it's a low hanging fruit and might give
a speed
>>>boost to many pipelines when run on the Flink runnner. I would
like to
>>>make this my first contribution to Beam, but as the guide [4]
says, I
>>>thought I'd ask here first to see if there a is a reason not to
do this.
>>>
>>>Only objection I can see, is that it might break existing pipelines
>>>which rely on the Flink runner saving them from not following the
>>>immutability guarantee. I see this as a small loss as they are
relying
>>>on an implementation detail of the Flink runner.
>>>
>>>I hope I have explained this adequately and eagerly away any
feedback :)
>>>
>>>Best regards,
>>>Teodor Spæren
>>>
>>>[1]: https://arxiv.org/abs/1907.08302
>>>[2]:
>>>https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
>>>[3]:
>>>https://github.com/apache/beam/blob/master/runners/flink/1.8/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java#L84
>>>[4]: https://beam.apache.org/contribute/
>>>
