Writing to Kafka from Apache Beam

This recipe shows how to write to Kafka with Apache Beam.

Getting ready

To install Apache Beam, follow the instructions at: https://beam.apache.org/get-started/quickstart-py/.

How to do it...

The following code shows how to write a Beam pipeline to write to Kafka. The example illustrates various options for configuring the Beam sink:

 
  PCollection<KV<Long, String>> kvColl = ...; 
  kvColl.apply(KafkaIO.write() 
       .withBootstrapServers("broker_1:9092, broker_2:9092") 
       .withTopic("destination-topic") 
 
       .withKeyCoder(BigEndianLongCoder.of()) 
       .withValueCoder(StringUtf8Coder.of()) 
 
       .updateProducerProperties( 
         ImmutableMap.of("compression.type", "gzip")) 
    ); 

How it works...

The KafkaIO sink supports writing key value pairs to a Kafka topic. To configure a Kafka sink, specify the Kafka bootstrap servers and the topic to write to.

The KafakIO allows setting most of the properties in the consumer configuration...