How to manually commit kafka offset after FileIO in apache beam?

user_4871474 0
Viewed 62

I have a FileIO writing a Pcollection<GenericRecord> to files and returns WriteFilesResult<DestinationT>.

I would like to create a DoFn after writing files to commit the offset of written records to kafka but since my offsets are stored in my GenericRecords I can no longer access them in the output of FileIO.

What is the best way to solve this ?

apache-beam-kafkaio apache-beam google-cloud-dataflow apache-kafka
1 Answers

For anyone interested, here is how I did:

  • manually groupbykey records by DestinationT
  • for each group I get the list of offsets and I create a new key EnrichedDestinationT + flatten the iterable
  • so the Pcollection before entering FileIO is PCollection<KV<EnrichedDestinationT, GenericRecord>>
  • in FileIO, the .by() becomes .by(KV::getKey) and the .via() becomes .via(Contextful.fn(KV::getValue), Contextful.fn(this::getSink))
user_48714740
Related