[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Integrating Stateful DoFns from the Python SDK

Type hints turn out to be not so predictable:

  p | beam.Impulse() \
    | beam.ParDo(MyCreate()).with_output_types(typehints.KV[K, V]) \
    | "statefulParDo" >> beam.ParDo(AddIndex())

2) DOES NOT (no KvCoder)
  p | beam.Create(inputs).with_output_types(typehints.KV[K, V]) \
    | "statefulParDo" >> beam.ParDo(AddIndex())

Do you know a way to make 2) work, i.e. set the KvCoder for the Create?

In the first example, the Create runs in a ParDo, in the second example
On 17.10.18 15:34, Maximilian Michels wrote:
Thanks Robert. I was able to get it working by adding this to the transform before my stateful DoFn:

   .with_output_types(typehints.KV[K, V])

For some reason `.with_input_types(typehints.KV[K, V])` on my stateful DoFn did not work.

Until we enforce KV during pipeline construction, we will have to throw an informative exception in the Runner.

On 17.10.18 15:03, Robert Bradshaw wrote:
Yes, we should be enforcing keyness (and use of KeyCoder with) stateful DoFns, similar to what we do for GBKs. See e.g. https://github.com/apache/beam/pull/6304#issuecomment-421935375

(This possibly relates to a long-standing issue that the coder inference should be moved up into construction, or at least before we pass the graph to the runner.)

On Wed, Oct 17, 2018 at 2:52 PM Maximilian Michels <mxm@xxxxxxxxxx <mailto:mxm@xxxxxxxxxx>> wrote:

    Hi everyone,

    While integrating portable state with the FlinkRunner, I hit a problem
    and wanted to get your opinion.

    Stateful DoFns require their input to be KV records. The reason for
    is that state is isolated by key. The (non-portable) FlinkRunner uses
    Flink's `keyBy(key)` construct to partition state by key [1].

    That works fine for portable Java pipelines where we enforce the `KV`
    class for Stateful DoFns. After running tests with the Python SDK, I
    came to the conclusion that tuples, e.g. `(key, value)` which are used
    for KV functionality, do not go through the KvCoder but are encoded
    using a byte array encoder.

    How do we infer the key in the Runner from an opaque sequence of bytes?     Should we also require the KvCoder for stateful DoFns in the Python SDK?