title: Deserialize Kafka AVRO messages using Apache Beam
body: <p>The main goal is the aggregate two Kafka topics, one compacted slow moving data and the other fast moving data which is received every second.</p>
  <p>I have been able to consume messages in simple scenarios such as a KV (Long,String) using something like:</p>
  <pre><code>PCollection&lt;KV&lt;Long,String&gt;&gt; input = p.apply(KafkaIO.&lt;Long, 
  String&gt;read()
  .withKeyDeserializer(LongDeserializer.class)
  .withValueDeserializer(StringDeserializer.class)
    
  PCollection&lt;String&gt; output = input.apply(Values.&lt;String&gt;create());
  </code></pre>
  <p>But this doesn’t seem to be the approach when you need to deserialize from AVRO. I have a KV(STRING, AVRO) which I need to consume.</p>
  <p>I attempted generating the Java Classes from the AVRO schema and then including them in the “apply” for example:</p>
  <pre><code>PCollection&lt;MyClass&gt; output = input.apply(Values.&lt;MyClass&gt;create());
  </code></pre>
  <p>But this didn’t seem to be the correct approach.</p>
  <p>Is there any documentation/examples anyone could point me to, so I could get an understanding as to how you would work with Kafka AVRO and Beam?</p>
  <p>I have updated my code:</p>
  <pre><code>import io.confluent.kafka.serializers.KafkaAvroDeserializer;
  import org.apache.beam.sdk.Pipeline;
  import org.apache.beam.sdk.coders.AvroCoder;
  import org.apache.beam.sdk.io.kafka.KafkaIO;
  import org.apache.beam.sdk.options.PipelineOptions;
  import org.apache.beam.sdk.options.PipelineOptionsFactory;
  import org.apache.beam.sdk.values.KV;
  import org.apache.beam.sdk.values.PCollection;
  import org.apache.kafka.common.serialization.LongDeserializer;
  
  public class Main {
  
  public static void main(String[] args) {
  
      PipelineOptions options = PipelineOptionsFactory.create();
  
      Pipeline p = Pipeline.create(options);
  
      PCollection&lt;KV&lt;Long, Myclass&gt;&gt; input = p.apply(KafkaIO.&lt;Long, String&gt;read()
              .withKeyDeserializer(LongDeserializer.class)
              .withValueDeserializerAndCoder(KafkaAvroDeserializer.class, AvroCoder.of(Myclass.class))
      );
  
      p.run();
  
  }
  }
  </code></pre>
  <hr />
  <pre><code>import org.apache.beam.sdk.coders.AvroCoder;
  import org.apache.beam.sdk.coders.DefaultCoder;
  
  @DefaultCoder(AvroCoder.class)
  public class Myclass{
  String name;
  String age;
  
  Myclass(){}
  Myclass(String n, String a) {
      this.name= n;
      this.age= a;
  }
  }
  </code></pre>
  <p>But i now get the following error</p>
  <pre><code>incompatible types: java.lang.Class &lt; io.confluent.kafka.serializers.KafkaAvroDeserializer &gt; cannot be converted to java.lang.Class &lt; ? extends org.apache.kafka.common.serialization.Deserializer &lt; java.lang.String &gt; &gt;
  </code></pre>
  <p>I must be importing the incorrect serializers?</p>