We have a batch GCP Dataflow job that is failing on a Reshuffle() step with the following error: ValueError: Error decoding input stream with coder WindowedValueCoder[TupleCoder[LengthPrefixCoder[DeterministicFastPrimitivesCoder], StateBackedIterableCoder[LengthPrefixCoder[FastPrimitivesCoder]]]] We have a step that generates n amount of lists, and then passes each list to the next PCollection for processing. Without a "Resfhuffle()" step, these two steps are fused together, and the lists from the first step are processed linearly, instead of in parallel. The Reshuffle() was added to get rid of the fusing and split up the lists between workers, but we are getting the error above during the "GroupByKey" step of the Reshuffle(). The ouput of Pcollection 1 is list objects, which the Reshuffle() is expected to pass to the next PCollection. I have tried changing the output of the first PCollection from a list to an object containing a list and received the same error. We've seen the same issue running Python3.11 and 3.12. We are using the DataflowRunner. The pipeline is set up as follows: with beam.Pipeline(options=beam_options) as p: (p | 'Get source data and split into lists of rows' >> beam.ParDo(GetDataAndSplit()) | "Reshuffle" >> beam.Reshuffle() | 'Process Lists from previous Pcollection' >> beam.ParDo(ProcessData()) ) Continue reading...
