Hi,
I wanted to get your opinion on something that I have been struggling
with. It is about the coders for state requests in portable pipelines.
In contrast to "classic" Beam, the Runner is not guaranteed to know
which coder is used by the SDK. If the SDK happens to use a standard
coder (also known as model coder), we will also have it available at the
Runner, i.e. if the Runner is written in one of the SDK languages (e.g.
Java). However, when we do not have a standard coder, we just treat the
data from the SDK as a blob and just pass it around as bytes.
Problem
=======
In the case of state requests which the SDK Harness authors to the
Runner, we would like for the key associated with the state request to
match the key of the element which led to initiating the state request.
Example:
Runner SDK Harness
------ -----------
KV["key","value"] --> Process Element
|
LookupState("key") <-- Request state of "key"
|
State["key"] --> Receive state
For stateful DoFns, the Runner partitions the data based on the key. In
Flink, this partitioning must not change during the lifetime of a
pipeline because the checkpointing otherwise breaks[0]. The key is
extracted from the element and stored encoded.
If we have a standard coder, it is basically the same as in the
"classic" Runner which takes the key and serializes it. However, when we
have an SDK-specific coder, we basically do not know how it encodes. So
far, we have been using the coder instantiated from the Proto, which is
basically a LengthPrefixCoder[ByteArrayCoder] or similar[1]. We have had
problems with this because the key encoding of Java SDK state requests
did not match the key encoding on the Runner side [2]. In an attempt to
fix those, it is now partly broken for portable Python pipelines.
Partly, because it "only" affects non-standard coders.
Non-standard coders yield the aforementioned
LengthPrefixCoder[ByteArrayCoder]. Now, following the usual encoding
scheme, we would simply encode the key using this coder. However, for
state requests, the Python SDK leaves out the length prefix for certain
coders, e.g. for primitives like int or byte. It is possible that one
coder uses a length prefix, while another doesn't. We have no way of
telling from the Runner side, if a length prefix has been used or not.
This results in the keys to not match on the Runner side and the
partitioning to be broken.
How to solve this?
==================
(1) Should this simply be fixed on the Python SDK side? One fix would be
to always append a length prefix to the key in state requests, even for
primitive coders like VarInt which do not use one.
OR
(2) Should the Runner detect that a non-standard coder is used? If so,
it should just pass the bytes from the SDK Harness and never make an
attempt to construct a coder based on the Proto.
Thinking about it now, it seems pretty obvious that (2) is the most
feasible way to avoid complications across all current and future SDKs
for key encodings. Still, it is odd that the Proto contains coder
information which is not usable.
What do you think?
Thanks,
Max
[0] It is possible to restart the pipeline and repartition the
checkpointed data.
[1]
https://github.com/apache/beam/blob/c39752af5391fe698a2b4f1489c187ddd4d604c0/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java#L682
[2] https://issues.apache.org/jira/browse/BEAM-8157
