On Sun, Jul 25, 2021 at 6:33 AM Jan Lukavský <je...@seznam.cz
<mailto:je...@seznam.cz>> wrote:
I'll start from the end.
I don't think we should be breaking language agnostic API
layers (for example, definition of model coders) just to
support "use_deprecated_read".
"Breaking" and "fixing" can only be a matter of the
definition of the object at hand. I don't think, that Coder
can be totally language agnostic - yes, the mapping between
serialized form and deserialized form can be _defined_ in a
language agnostic way, but must be_implemented_ in a
specific language. If we choose the implementing language,
what makes us treat SDK-specific coders defined by the SDK
of the same language as "unknown"? It is only our decision,
that seems to have no practical benefits.
In general, language-neutral APIs and protocols are a key
feature of portable Beam. See here:
https://beam.apache.org/roadmap/portability/
<https://beam.apache.org/roadmap/portability/>
(I did not look into all the old discussions and votes related
to this but I'm sure they are there)
Moreover, including SDK-specific coders into supported
coders of the SDK runner construction counterpart (that is,
runner core-construction-java for Java SDK) is a necessary
prerequisite for unifying "classical" and "portable"
runners, because the runner needs to understand *all* SDK
coders so that it can _inline_ the complete Pipeline (if the
Pipeline SDK has the same language as the runner), instead
of running it through SDK harness. This need therefore is
not specific to supporting use_deprecated_read, but is a
generic requirement, which only has the first manifestation
in the support of a transform not supported by SDK harness.
I think "use_deprecated_read" should be considered a
stop-gap measure for Flink (and Spark ?) till we have
proper support for SDF. In fact I don't think an arbitrary
portable runner can support "use_deprecated_read" due to
the following.
There seems to be nothing special about Flink regarding the
support of primitive Read. I think any Java native runner
can implement it pretty much the same way as Flink does. The
question is if any other runner might want to do that. The
problem with Flink is that
Not all runners are implemented using Java. For example, the
portable DirectRunner (FnAPI runner) is implemented using Python
and Dataflow is implemented using C++. Such runners will not be
able to do this.
1) portable SDF seems not to work [1]
2) even classical Flink runner has still issues with SDF -
there are reports of watermark being stuck when reading data
via SDF, this gets resolved using use_deprecated_read
3) Flink actually does not have any benefits from SDF,
because it cannot make use of the dynamic splitting, so this
actually brings only implementation burden without any
practical benefit
Similarly, I think there were previous discussions related to
using SDF as the source framework for portable runners.
I understand that there are some bugs related to SDF and
portable Flink currently. How much work do you think is needed
here ? Will it be better to focus our efforts on fixing
remaining issues for SDF and portable runners instead of
supporting "use_deprecated_read" for that path ? Note that I'm
fine with fixing any issues related to "use_deprecated_read" for
classic (non-portable) Flink but I think you are trying to use
x-lang hence probably need portable Flink.
Thanks,
Cham
I think that we should reiterate on the decision of
deprecating Read - if we can implement it via SDF, what is
the reason to forbid a runner to make use of a simpler
implementation? The expansion of Read might be runner
dependent, that is something we do all the time, or am I
missing something?
Jan
[1] https://issues.apache.org/jira/browse/BEAM-10940
<https://issues.apache.org/jira/browse/BEAM-10940>
On 7/25/21 1:38 AM, Chamikara Jayalath wrote:
I think we might be going down a bit of a rabbit hole with
the support for "use_deprecated_read" for portable Flink :)
I think "use_deprecated_read" should be considered a
stop-gap measure for Flink (and Spark ?) till we have
proper support for SDF. In fact I don't think an arbitrary
portable runner can support "use_deprecated_read" due to
the following.
(1) SDK Harness is not aware of
BoundedSource/UnboundedSource. Only source framework SDK
Harness is aware of is SDF.
(2) Invoking BoundedSource/UnboundedSource is not a part of
the Fn API
(3) A non-Java Beam portable runner will probably not be
able to directly invoke legacy Read transforms similar to
the way Flink does today.
I don't think we should be breaking language agnostic API
layers (for example, definition of model coders) just to
support "use_deprecated_read".
Thanks,
Cham
On Sat, Jul 24, 2021 at 11:50 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
On 7/24/21 12:34 AM, Robert Bradshaw wrote:
> On Thu, Jul 22, 2021 at 10:20 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>> Hi,
>>
>> this was a ride. But I managed to get that working.
I'd like to discuss two points, though:
>>
>> a) I had to push Java coders to ModelCoders for
Java (which makes sense to me, but is that correct?).
See [1]. It is needed so that the Read transform
(executed directly in TaskManager) can correctly
communicate with Java SDK harness using custom coders
(which is tested here [2]).
> I think the intent was that ModelCoders represent the
set of
> language-agnostic in the model, though I have to
admit I've always
> been a bit fuzzy on when a coder must or must not be
in that list.
I think that this definition works as long, as runner
does not itself
interfere with the Pipeline. Once the runner starts (by
itself, not via
SdkHarnessClient) producing data, it starts to be part
of the
environment, and therefore it should understand its own
Coders. I'd
propose the definition of "model coders" to be Coders
that the SDK is
able to understand, which then works naturally for the
ModelCoders
located in "core-construction-java", that it should
understand Javs SDK
Coders.
>
>> b) I'd strongly prefer if we moved the handling of
use_deprecated_read from outside of the Read PTransform
directly into expand method, see [3]. Though this is
not needed for the Read on Flink to work, it seems cleaner.
>>
>> WDYT?
> The default value of use_deprecated_read should
depend on the runner
> (e.g. some runners don't work well with it, others
require it). As
> such should not be visible to the PTransform's expand.
I think we should know what is the expected outcome. If
a runner does
not support primitive Read (and therefore
use_deprecated_read), what
should we do, if we have such experiment set? Should
the Pipeline fail,
or should it be silently ignored? I think that we
should fail, because
user expects something that cannot be fulfilled.
Therefore, we have two
options - handling the experiment explicitly in runners
that do not
support it, or handle it explicitly in all cases (both
supported and
unsupported). The latter case is when we force runners
to call explicit
conversion method (convertPrimitiveRead....). Every
runner that does not
support primitive Read must handle the experiment
either way, because
otherwise the experiment would be simply silently
ignored, which is not
exactly user-friendly.
>
>> Jan
>>
>> [1]
https://github.com/apache/beam/pull/15181/commits/394ddc3fdbaacc805d8f7ce02ad2698953f34375
<https://github.com/apache/beam/pull/15181/commits/394ddc3fdbaacc805d8f7ce02ad2698953f34375>
>>
>> [2]
https://github.com/apache/beam/pull/15181/files#diff-b1ec58edff6c096481ff336f6fc96e7ba5bcb740dff56c72606ff4f8f0bf85f3R201
<https://github.com/apache/beam/pull/15181/files#diff-b1ec58edff6c096481ff336f6fc96e7ba5bcb740dff56c72606ff4f8f0bf85f3R201>
>>
>> [3]
https://github.com/apache/beam/pull/15181/commits/f1d3fd0217e5513995a72e92f68fe3d1d665c5bb
<https://github.com/apache/beam/pull/15181/commits/f1d3fd0217e5513995a72e92f68fe3d1d665c5bb>
>>
>> On 7/18/21 6:29 PM, Jan Lukavský wrote:
>>
>> Hi,
>>
>> I was debugging the issue and it relates to pipeline
fusion - it seems that the primitive Read transform
gets fused and then is 'missing' as source. I'm a
little lost in the code, but the most strange parts are
that:
>>
>> a) I tried to reject fusion of primitive Read by
adding GreedyPCollectionFusers::cannotFuse for
PTransformTranslation.READ_TRANSFORM_URN to
GreedyPCollectionFusers.URN_FUSIBILITY_CHECKERS, but
that didn't change the exception
>>
>> b) I tried adding Reshuffle.viaRandomKey between
Read and PAssert, but that didn't change it either
>>
>> c) when I run portable Pipeline with
use_deprecated_read on Flink it actually runs (though
it fails when it actually reads any data, but if the
input is empty, the job runs), so it does not hit the
same issue, which is a mystery to me
>>
>> If anyone has any pointers that I can investigate,
I'd be really grateful.
>>
>> Thanks in advance,
>>
>> Jan
>>
>>
>>
>> On 7/16/21 2:00 PM, Jan Lukavský wrote:
>>
>> Hi,
>>
>> I hit another issue with the portable Flink runner.
Long story short - reading from Kafka is not working in
portable Flink. After solving issues with expansion
service configuration (ability to add
use_deprecated_read) option, because flink portable
runner has issues with SDF [1], [2]. After being able
to inject the use_deprecated_read into expansion
service I was able to get an execution DAG that has the
UnboundedSource, but then more and more issues appeared
(probably related to missing LengthPrefixCoder
somewhere - maybe at the output from the primitive
Read). I wanted to create a test for it and I found
out, that there actually is ReadSourcePortableTest in
FlinkRunner, but _it tests nothing_. The problem is
that Read is transformed to SDF, so this test tests the
SDF, not the Read transform. As a result, the Read
transform does not work.
>>
>> I tried using
convertReadBasedSplittableDoFnsToPrimitiveReads so that
I could make the test fail and debug that, but I got into
>>
>> java.lang.IllegalArgumentException: PCollectionNodes
[PCollectionNode{id=PAssert$0/GroupGlobally/ParDo(ToSingletonIterables)/ParMultiDo(ToSingletonIterables).output,
PCollection=unique_name:
"PAssert$0/GroupGlobally/ParDo(ToSingletonIterables)/ParMultiDo(ToSingletonIterables).output"
>> coder_id: "IterableCoder"
>> is_bounded: BOUNDED
>> windowing_strategy_id:
"WindowingStrategy(GlobalWindows)"
>> }] were consumed but never produced
>>
>>
>> which gave me the last knock-out. :)
>>
>> My current impression is that starting from Beam
2.25.0, portable FlinkRunner is not able to read from
Kafka. Could someone give me a hint about what is wrong
with using
convertReadBasedSplittableDoFnsToPrimitiveReads in the
test [3]?
>>
>> Jan
>>
>> [1] https://issues.apache.org/jira/browse/BEAM-11991
<https://issues.apache.org/jira/browse/BEAM-11991>
>>
>> [2] https://issues.apache.org/jira/browse/BEAM-11998
<https://issues.apache.org/jira/browse/BEAM-11998>
>>
>> [3] https://github.com/apache/beam/pull/15181
<https://github.com/apache/beam/pull/15181>
