Deserializers

<dependency>
    <groupId>com.landoop</groupId>
    <artifactId>lenses-serde</artifactId>
    <version>1.0.3</version>
</dependency>

public class LatLngSerde implements Serde {
  private Schema schema = SchemaBuilder.builder()
                          .record("lat_lng")
                          .fields()
                          .requiredDouble("lat")
                          .requiredDouble("lng")
                          .endRecord();

  @Override
  public Schema getSchema() {
    return schema;
  }

  @Override
  public Serializer serializer(Properties properties) {
    return null;
  }

  @Override
  public Deserializer deserializer(Properties properties) {
    return new Deserializer() {
      @Override
      public GenericRecord deserialize(byte[] bytes) throws IOException {
        String data = new String(bytes);
        String[] tokens = data.split(":");
        double lat = Double.parseDouble(tokens[0]);
        double lng = Double.parseDouble(tokens[1]);

        GenericRecord record = new GenericData.Record(schema);
        record.put("lat", lat);
        record.put("lng", lng);
        return record;
      }

      @Override
      public void close() throws IOException {
      }
    };
  }
}