Apache Beam, как распаковать ключ и значения в процессе dofn ⇐ Python

[Anonymous]

Ниже приведен мой код Python и моя цель — найти различия между pcollections dataset1 и dataset2. Печать элемента работает. Однако, когда я пытаюсь распаковать элемент как ключ, значение, которое он выдает, и ошибка - невозможно распаковать неитерируемый объект NoneType
, элемент печатается, как показано ниже
('data1a', {'dataset1': ['data1a'], 'dataset2': []})
None
('data1b', {'dataset1': ['data1b'], 'dataset2': ['data1b']})
None

мой код показан ниже, и я закомментировал часть ключа и значения
import apache_beam as beam

def make_kv_pair(x):
""" double the record. is this a good practice"""
return (x, x)
with beam.Pipeline() as p:

class FilterDoFn(beam.DoFn):
def process(self, element):
print(element)
""" why the below doesnt work"""
#key,value = element

dataset1 = ['data1a', 'data1b', 'data1c', 'data1d']
dataset2 = ['data2a', 'data2b', 'data1b', 'data2d']
output_types = ['changed', 'added', 'deleted', 'unchanged']
dataset1_pcoll = (p | 'Read Dataset 1' >> beam.Create(dataset1) | 'doubledata1' >> beam.Map(make_kv_pair))
dataset2_pcoll = (p | 'Read Dataset 2' >> beam.Create(dataset2) | 'doubledata2' >> beam.Map(make_kv_pair))

combined_data_collection = (
{
'dataset1': dataset1_pcoll,
'dataset2': dataset2_pcoll,
}
| 'Group datasets' >> beam.CoGroupByKey()
| 'prinnting' >> beam.Map(print)
)

key_collections = (combined_data_collection
| beam.ParDo(FilterDoFn()).with_outputs(*output_types))



Подробнее здесь: https://stackoverflow.com/questions/791 ... fn-process