在流式流水线中组合多个边输入时数据流失败

Question

我已经用Python SDK（Apache Beam Python 3.7 SDK 2.19.0）构建了一个窗口流数据流管道。初始数据的表示如下：

| Phone Number | Call length |
|--------------|-------------|
| 1234         | 6           |
| 1234         | 2           |
| 5678         | 5           |

其思想是找出给定窗口中每行号码的平均通话长度。数据作为CSV的行从pub/sub中读取，我向所有行添加一个与该数字的平均调用长度相对应的值：

| Phone Number | Call length | mean call length |
|--------------|-------------|------------------|
| 1234         | 6           | 4                |
| 1234         | 2           | 4                |
| 5678         | 5           | 5                |

我使用以下管道：

    with beam.Pipeline(options=pipeline_options) as pipeline:
        calls = (pipeline
             | 'Read PubSub Messages' >> beam.io.ReadFromPubSub(subscription=input_sub)
             | 'byte_to_string' >> beam.Map(lambda x: x.decode("utf-8"))
             | 'windows' >> beam.WindowInto(window.FixedWindows(10))   
            )

        mean_call_length = (calls
             | 'call_length_for_number' >> beam.ParDo(get_list_of_pairs_of_tuples(),'number','call_length')
             |  'mean_call_length_per_number' >> beam.combiners.Mean.PerKey()
            )

        recombine = (calls
              | 'Create dictionary from raw string' >> beam.ParDo(SplitToDict())
              |  'Add mean' >> beam.FlatMap(combine_calcs,pvalue.AsList(mean_call_length))
              | 'encode to bytes' >> beam.Map(lambda x: str(x).encode())
              | 'write to output topic' >> beam.io.WriteToPubSub(topic=output_topic)
            )

Caused by: org.apache.beam.sdk.coders.CoderException: java.io.EOFException 

有什么想法吗？

Answer 1

我通过更改管道将平均值计算的结果转换为整数来解决这个问题：

...
mean_call_length = (calls
             | 'call_length_for_number' >> beam.ParDo(get_list_of_pairs_of_tuples(),'number','call_length')
             |  'mean_call_length_per_number' >> beam.combiners.Mean.PerKey())
             | 'convert_mean_to_int' >> beam.Map(lambda elem: (elem[0],int(elem[1])))
...

Python SDK和底层Java代码之间似乎存在一些键入问题；Java代码似乎期望元素[1]低于一定的字节数，如果通过Python SDK提交浮点数，则超过这个字节数。