• 18 years with Object Partners
• Technologies
  • Kafka (Core, Streams, KSQL, Connectors)
  • Spring / Spring Boot
  • Cassandra
  • Hibernate / JPA
  • JMS / IBMMQ / RabbitMQ
• https://objectpartners.com/author/nbuesing/
• https://github.com/nbuesing

Object Partners

• www.objectpartners.com (not opi.com)
• 20+ year history
• history of successful enterprise software development
• Chris Spurgat / President, CEO, Co-Founder
  • https://objectpartners.com/author/cspurgat/

Overview

• Confluent Development CLI
• 1 Connector
  • Custom OpenSky Connector
• 3 Microservices
  • Geolocation RESTful API
  • Kafka Steams
  • Kafka Consumer & D3 Web Application
• KSQL
  • Custom Library

AVRO

Location

{
  "namespace": "com.objectpartners.buesing.avro",
  "name": "Location",
  "type": "record",
  "fields": [
    {
      "name": "latitude",
      "type": "double"
    },
    {
      "name": "longitude",
      "type": "double"
    }
  ]
}

AVRO

Record

{
  "namespace": "com.objectpartners.buesing.avro",
  "name": "Record",
  "type": "record",
  "fields": [
    {
      "name": "origin",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "aircraft",
      "type": "com.objectpartners.buesing.avro.Aircraft"
    },
    {
      "name": "location",
      "type": "com.objectpartners.buesing.avro.Location"
    },
    {
      "name": "altitude",
      "type": ["null", "double"],
      "default": null
    },
    {
      "name": "onGround",
      "type": "boolean"
    },
    {
      "name": "velocity",
      "type": ["null", "double"],
      "default": null
    },
    {
      "name": "verticalRate",
      "type": ["null", "double"],
      "default": null
    }
  ]
}

AVRO

Distance

{
  "namespace": "com.objectpartners.buesing.avro",
  "name": "Distance",
  "type": "record",
  "fields": [
    {
      "name": "red",
      "type": "com.objectpartners.buesing.avro.Record"
    },
    {
      "name": "blue",
      "type": "com.objectpartners.buesing.avro.Record"
    },
    {
      "name": "distance",
      "type": "double"
    }
  ]
}

AVRO

Nearest Airport

{
  "namespace": "com.objectpartners.buesing.avro",
  "name": "NearestAirport",
  "type": "record",
  "fields": [
    {
      "name": "airport",
      "type": "string"
    },
    {
      "name": "latitude",
      "type": "double"
    },
    {
      "name": "longitude",
      "type": "double"
    }
  ]
}

AVRO

Count

{
  "namespace": "com.objectpartners.buesing.avro",
  "name": "Count",
  "type": "record",
  "fields": [
    {
      "name": "label",
      "type": "string"
    },
    {
      "name": "value",
      "type": "int"
    }
  ]
}

Streams - blue

@Bean
public KStream<String, Record> blue() {
  return streamsBuilder
    .stream("blue",
      Consumed.with(Serdes.String(), recordSerde));
}

Streams - nearest airport

@Bean
public KStream<String, NearestAirport> nearestAirport() {
  return red().map((key, value) -> {
    String airport = geolocation.closestAirport(
      value.getLocation().getLatitude(),
      value.getLocation().getLongitude()
    ).getCode();

    return KeyValue.pair(
      key,
      new NearestAirport(
        airport,
        value.getLocation().getLatitude(),
        value.getLocation().getLongitude()
      )
    );
  }).through(
      "red.nearest.airport",
      Produced.with(Serdes.String(), nearestAirportSerde)
  );
}

Streams - count

@Bean
public KStream<String, Count> nearestAirportCount() {
  KStream<String, Count> bean = nearestAirport()
    .selectKey((key, value) -> value.getAirport())
    .groupByKey()
    .windowedBy(TimeWindows.of(WINDOW))
    .aggregate(
      () -> 0,
      (key, value, aggregate) -> aggregate + 1,
      Materialized.with(Serdes.String(), Serdes.Integer())
    ).toStream((wk, v) -> wk.key())
    .mapValues(Count::new)
    .through("red.nearest.airport.count", Produced.with(Serdes.String(), countSerde)
  );
  return bean;
}

KSQL - blue

create stream blue with (kafka_topic='blue', value_format='avro');

@Bean
public KStream<String, Record> blue() {
  return streamsBuilder
    .stream("blue",
      Consumed.with(Serdes.String(), recordSerde));
}

Streams - nearest airport

create stream \
  ksql_red_nearest_airport with (PARTITIONS=8) \
  as select \
      aircraft->transponder transponder, \
      closestAirport(location->latitude, location->longitude) as airport, \
      location \
    from red \
  partition by transponder;

@Bean
public KStream<String, NearestAirport> nearestAirport() {
  return red().map((key, value) -> {
    String airport = geolocation.closestAirport(
      value.getLocation().getLatitude(),
      value.getLocation().getLongitude()
    ).getCode();

    return KeyValue.pair(
      key,
      new NearestAirport(
        airport,
        value.getLocation().getLatitude(),
        value.getLocation().getLongitude()
      )
    );
  }).through(
      "red.nearest.airport",
      Produced.with(Serdes.String(), nearestAirportSerde)
  );
}

Streams - count

create table \
  ksql_red_nearest_airport_count with (PARTITIONS=8) \
  as select \
      airport, \
      count(*) as count \
    from ksql_red_nearest_airport window tumbling (size 4 hours) \
  group by airport;

@Bean
public KStream<String, Count> nearestAirportCount() {
  KStream<String, Count> bean = nearestAirport()
    .selectKey((key, value) -> value.getAirport())
    .groupByKey()
    .windowedBy(TimeWindows.of(WINDOW))
    .aggregate(
      () -> 0,
      (key, value, aggregate) -> aggregate + 1,
      Materialized.with(Serdes.String(), Serdes.Integer())
    ).toStream((wk, v) -> wk.key())
    .mapValues(Count::new)
    .through("red.nearest.airport.count", Produced.with(Serdes.String(), countSerde)
  );
  return bean;
}

KSQL

SET 'auto.offset.reset' = 'earliest';

create stream red with (kafka_topic='red', value_format='avro');

create stream blue with (kafka_topic='blue', value_format='avro');

create stream \
  ksql_red_nearest_airport with (PARTITIONS=8) \
  as select \
      aircraft->transponder transponder, \
      closestAirport(location->latitude, location->longitude) as airport, \
      location \
    from red \
  partition by transponder;

create table \
  ksql_red_nearest_airport_count with (PARTITIONS=8) \
  as select \
      airport, \
      count(*) as count \
    from ksql_red_nearest_airport window tumbling (size 4 hours) \
  group by airport;

Topology : KSQL : nearest airport

   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [red])
      --> KSTREAM-MAPVALUES-0000000001
    Processor: KSTREAM-MAPVALUES-0000000001 (stores: [])
      --> KSTREAM-TRANSFORMVALUES-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KSTREAM-TRANSFORMVALUES-0000000002 (stores: [])
      --> KSTREAM-MAPVALUES-0000000003
      <-- KSTREAM-MAPVALUES-0000000001
    Processor: KSTREAM-MAPVALUES-0000000003 (stores: [])
      --> KSTREAM-FILTER-0000000004
      <-- KSTREAM-TRANSFORMVALUES-0000000002
    Processor: KSTREAM-FILTER-0000000004 (stores: [])
      --> KSTREAM-KEY-SELECT-0000000005
      <-- KSTREAM-MAPVALUES-0000000003
    Processor: KSTREAM-KEY-SELECT-0000000005 (stores: [])
      --> KSTREAM-MAPVALUES-0000000006
      <-- KSTREAM-FILTER-0000000004
    Processor: KSTREAM-MAPVALUES-0000000006 (stores: [])
      --> KSTREAM-MAPVALUES-0000000007
      <-- KSTREAM-KEY-SELECT-0000000005
    Processor: KSTREAM-MAPVALUES-0000000007 (stores: [])
      --> KSTREAM-SINK-0000000008
      <-- KSTREAM-MAPVALUES-0000000006
    Sink: KSTREAM-SINK-0000000008 (topic: KSQL_RED_NEAREST_AIRPORT)
      <-- KSTREAM-MAPVALUES-0000000007

User Defined Function

@UdfDescription(name = "closestAirport", description = "return airport code for closest airport.")
public class ClosestAirport {
    private static final String HOST = "http://localhost:9080";
    private Geolocation geolocation = Feign.builder()
            .options(new Request.Options(200, 200))
            .encoder(new JacksonEncoder())
            .decoder(new JacksonDecoder())
            .target(Geolocation.class, HOST);
    @Udf(description = "find closest airport to given location.")
    public String closestAirport(final Double latitude, final Double longitude) {
        return geolocation.closestAirport(latitude, longitude).getCode();
    }
}

Insert Into

create stream blueBucket \
  as select \
    bucketLocation(location->latitude, location->longitude, 3.0) as bucket, \
    aircraft->transponder as transponder, aircraft->callsign as callsign, \
    location->latitude as latitude, location->longitude as longitude \
  from blue partition by bucket;

create stream blueBucket_w \
  as select \
    bucketLocation(location->latitude, location->longitude, 3.0, 'w') as bucket, \
    aircraft->transponder as transponder, aircraft->callsign as callsign, \
    location->latitude as latitude, location->longitude as longitude \
  from blue partition by bucket;

insert into blueBucket select * from blueBucket_w;

Stream-Stream Join

create stream redBlueJoin with (PARTITIONS=8) \
  as select \
    redBucket.transponder rt, \
    redBucket.latitude as rlat, \
    redBucket.longitude as rlong, \
    blueBucket.transponder as bt, \
    blueBucket.latitude as blat, \
    blueBucket.longitude as blong \
  from redBucket \
  inner join blueBucket within 4 hours \
    on redBucket.bucket = blueBucket.bucket;

KSQL

• KSQL != SQL (just like CQL != SQL)
• Great way to learn/experiment about streams
• Printing Topics & AVRO
• KSQL 2.0 feature == Production Ready
• Roadmap
  • constantly improving
  • fully support expressions (e.g. in partitioning)

Resources

• OPI Blog
• Confluent Blog
• Confluent You Tube Channel
• Confluent Streaming Audio Podcast
• Kafka Streams Examples