Kafka as a Platform: the Ecosystem from the Ground Up

Kafka as a Platform: the Ecosystem from the Ground Up https://events.confluent.io/meetups Robin Moffatt @rmoff

$ whoami • Robin Moffatt (@rmoff) • Senior Developer Advocate at Confluent (Apache Kafka, not Wikis 😉) • Working in data & analytics since 2001 • Oracle ACE Director (Alumnus) http://rmoff.dev/talks · http://rmoff.dev/blog · http://rmoff.dev/youtube @rmoff | #ConfluentVUG | @confluentinc

EVENTS @rmoff | #ConfluentVUG | @confluentinc

EVENTS n o i t a c fi i Not • e g n a h c e t a t S •

Human generated events A Sale A Stock movement @rmoff | #ConfluentVUG | @confluentinc

Machine generated events IoT Networking Applications @rmoff | #ConfluentVUG | @confluentinc

EVENTS are EVERYWHERE @rmoff | #ConfluentVUG | @confluentinc

EVENTS y r e v ^ are POWERFUL @rmoff | #ConfluentVUG | @confluentinc

K V

LOG @rmoff | #ConfluentVUG | @confluentinc

K V

Immutable Event Log Old New Events are added at the end of the log @rmoff | #ConfluentVUG | @confluentinc

TOPICS @rmoff | #ConfluentVUG | @confluentinc

Topics Clicks Orders Customers Topics are similar in concept to tables in a database @rmoff | #ConfluentVUG | @confluentinc

PARTITIONS @rmoff | #ConfluentVUG | @confluentinc

Partitions Clicks p0 P1 P2 Messages are guaranteed to be strictly ordered within a partition @rmoff | #ConfluentVUG | @confluentinc

PUB / SUB @rmoff | #ConfluentVUG | @confluentinc

Producing data Old New Messages are added at the end of the log @rmoff | #ConfluentVUG | @confluentinc

partition 0 … partition 1 producer … partition 2 … Partitioned Topic

try (KafkaProducer<String, Payment> producer = new KafkaProducer<String, Payment>(props)) { for (long i = 0; i < 10; i++) { final String orderId = “id” + Long.toString(i); final Payment payment = new Payment(orderId, 1000.00d); final ProducerRecord<String, Payment> record = new ProducerRecord<String, Payment>(“transactions”, payment.getId().toString(), payment); producer.send(record); } } catch (final InterruptedException e) { e.printStackTrace(); }

Producing to Kafka - No Key Time Partition 1 Partition 2 Partition 3 Messages will be produced in a round robin fashion Partition 4 @rmoff | #ConfluentVUG | @confluentinc

Producing to Kafka - With Key Time Partition 1 A Partition 2 B hash(key) % numPartitions = N Partition 3 C Partition 4 D @rmoff | #ConfluentVUG | @confluentinc

Producers partition 0 … partition 1 producer … partition 2 … Partitioned Topic • A client application • Puts messages into topics • Handles partitioning, network protocol • Java, Go, .NET, C/C++, Python • Also every other language

PUB / SUB @rmoff | #ConfluentVUG | @confluentinc

Consuming data - access is only sequential Read to offset & scan Old New @rmoff | #ConfluentVUG | @confluentinc

Consumers have a position of their own Old New Sally is here @rmoff | Scan #ConfluentVUG | @confluentinc

Consumers have a position of their own Old New Fred is here Sally is here Scan @rmoff | Scan #ConfluentVUG | @confluentinc

Consumers have a position of their own Rick is here Scan Old New Fred is here Sally is here Scan @rmoff | Scan #ConfluentVUG | @confluentinc

try (final KafkaConsumer<String, Payment> consumer = new KafkaConsumer<>(props)) { consumer.subscribe(Collections.singletonList(TOPIC)); while (true) { ConsumerRecords<String, Payment> records = consumer.poll(100); for (ConsumerRecord<String, Payment> record : records) { String key = record.key(); Payment value = record.value(); System.out.printf(“key = %s, value = %s%n”, key, value); } } }

partition 0 … partition 1 … partition 2 … Partitioned Topic consumer A

partition 0 … consumer A partition 1 … partition 2 … Partitioned Topic consumer B

Consuming From Kafka - Multiple Consumers C1 Partition 1 Partition 2 Partition 3 C2 Partition 4 @rmoff | #ConfluentVUG | @confluentinc

Consuming From Kafka - Grouped Consumers CC1 1 CC1 1 Partition 1 Partition 2 Partition 3 C2 Partition 4 @rmoff | #ConfluentVUG | @confluentinc

CONSUMERS CONSUMER GROUP COORDINATOR CONSUMER GROUP

Consuming From Kafka - Grouped Consumers Partition 1 Partition 2 Partition 3 C1 C2 C3 C4 Partition 4 @rmoff | #ConfluentVUG | @confluentinc

Consuming From Kafka - Grouped Consumers Partition 1 Partition 2 Partition 3 C1 C2 C3 3 Partition 4 @rmoff | #ConfluentVUG | @confluentinc

Consuming From Kafka - Grouped Consumers Partition 1 C1 Partition 2 Partition 3 C2 C3 Partition 4 @rmoff | #ConfluentVUG | @confluentinc

Consumers partition 0 … partition 1 … consumer A consumer A consumer A partition 2 … Partitioned Topic consumer B • A client application • Reads messages from topics • Horizontally, elastically scalable (if stateless) • Java, Go, .NET, C/C++, Python, everything else

BROKERS and REPLICATION @rmoff | #ConfluentVUG | @confluentinc

Leader Partition Leadership and Replication Follower Partition 1 Partition 2 Partition 3 Partition 4 Broker 1 Broker 2 @rmoff Broker 3 | #ConfluentVUG | @confluentinc

Leader Partition Leadership and Replication Follower Partition 1 Partition 1 Partition 1 Partition 2 Partition 2 Partition 2 Partition 3 Partition 3 Partition 3 Partition 4 Partition 4 Partition 4 Broker 1 Broker 2 Broker 3 @rmoff | #ConfluentVUG | @confluentinc

Photo by Raoul Droog on Unsplas DEMO @rmoff | #ConfluentVUG | @confluentinc

So far, this is Pretty good @rmoff | #ConfluentVUG | @confluentinc

So far, this is Pretty good but I’ve not finished yet… @rmoff | #ConfluentVUG | @confluentinc

Streaming Pipelines Amazon S3 RDBMS HDFS @rmoff | #ConfluentVUG | @confluentinc

Evolve processing from old systems to new Existing New App <x> App RDBMS @rmoff | #ConfluentVUG | @confluentinc

Streaming Integration with Kafka Connect syslog Sources Kafka Connect @rmoff | Kafka Brokers #ConfluentVUG | @confluentinc

Streaming Integration with Kafka Connect Amazon Sinks Google Kafka Connect @rmoff | Kafka Brokers #ConfluentVUG | @confluentinc

Streaming Integration with Kafka Connect Amazon syslog Google Kafka Connect @rmoff | Kafka Brokers #ConfluentVUG | @confluentinc

Look Ma, No Code! { “connector.class”: “io.confluent.connect.jdbc.JdbcSourceConnector”, “connection.url”: “jdbc:mysql://asgard:3306/demo”, “table.whitelist”: “sales,orders,customers” } @rmoff | #ConfluentVUG | @confluentinc

Extensible Connector Transform(s) @rmoff | Converter #ConfluentVUG | @confluentinc

hub.confluent.io @rmoff | #ConfluentVUG | @confluentinc

Single Kafka Connect node S3 Task #1 JDBC Task #1 JDBC Task #2 Kafka Connect cluster Worker Worker Offsets Config Status @rmoff | #ConfluentVUG | @confluentinc

Kafka Connect - scalable and fault-tolerant S3 Task #1 JDBC Task #1 Kafka Connect cluster JDBC Task #2 Worker Worker Offsets Config Status @rmoff | #ConfluentVUG | @confluentinc

Automatic fault tolerance S3 Task #1 JDBC Task #1 JDBC Task #2 Kafka Connect cluster Worker Worker Offsets Config Status @rmoff | #ConfluentVUG | @confluentinc

K V

K V ? s i h t s ’ t a h w … t i a W

How do you serialise your data? JSON Avro Protobuf Schema JSON CSV @rmoff | #ConfluentVUG | @confluentinc

APIs are contracts between services {user_id: 53, address: “2 Elm st.”} Profile service Quote service {user_id: 53, quote: 580} @rmoff | #ConfluentVUG | @confluentinc

But not all services {user_id: 53, address: “2 Elm st.”} Profile service Quote service {user_id: 53, quote: 580} @rmoff | #ConfluentVUG | @confluentinc

And naturally… {user_id: 53, address: “2 Elm st.”} Profile service Quote service Profile database @rmoff Stream processing | #ConfluentVUG | @confluentinc

Schemas are about how teams work together {user_id: 53, timestamp: 1497842472} new Date(timestamp) Profile service Quote service Profile database Stream processing create table ( user_id number, timestamp number) @rmoff | #ConfluentVUG | @confluentinc

Things change… {user_id: 53, timestamp: “June 28, 2017 4:00pm”} Profile service Quote service Profile database @rmoff Stream processing | #ConfluentVUG | @confluentinc

Moving fast and breaking things {user_id: 53, timestamp: “June 28, 2017 4:00pm”} Profile service Quote service Profile database @rmoff Stream processing | #ConfluentVUG | @confluentinc

Lack of schemas – Coupling teams and services 2001 2001 Citrus Heights-Sunrise Blvd Citrus_Hghts 60670001 3400293 34 SAC Sacramento SV Sacramento Valley SAC Sacramento County APCD SMA8 Sacramento Metropolitan Area CA 6920 Sacramento 28 6920 13588 7400 Sunrise Blvd 95610 38 41 56 38.6988889 121 16 15.98999977 -121.271111 10 4284781 650345 52 @rmoff | #ConfluentVUG | @confluentinc

Serialisation & Schemas JSON Avro Protobuf Schema JSON CSV @rmoff | #ConfluentVUG | @confluentinc

Serialisation & Schemas JSON Avro Protobuf Schema JSON CSV 👍 👍 👍 😬 https://qconnewyork.com/system/files/presentation-slides/qcon_17_-_schemas_and_apis.pdf @rmoff | #ConfluentVUG | @confluentinc

Schemas Schema Registry Topic producer … consumer

n a y l l Actua ! c i p o t l a n r e t in Schemas Schema Registry Topic producer … consumer

Producers contain serializers props.put(“key.serializer”, “org.apache.kafka.serializers.StringSerializer”); props.put(“value.serializer”, “io.confluent.kafka.serializers.KafkaAvroSerializer”); props.put(“schema.registry.url”, “http://schema-registry:8081”); … producer<String, LogLine> producer = new KafkaProducer<String, LogLine>(props); @rmoff | #ConfluentVUG | @confluentinc

partition 0 consumer A … consumer A partition 1 … consumer A partition 2 … consumer B Partitioned Topic @rmoff | #ConfluentVUG | @confluentinc

consumer A consumer A consumer A @rmoff | #ConfluentVUG | @confluentinc

} “reading_ts”: “2020-02-14T12:19:27Z”, “sensor_id”: “aa-101”, “production_line”: “w01”, “widget_type”: “acme94”, “temp_celcius”: 23, “widget_weight_g”: 100 Photo by Franck V. on Unsplash { @rmoff | #ConfluentVUG | @confluentinc

Streams of events Time @rmoff | #ConfluentVUG | @confluentinc

Stream Processing Stream: widgets Stream: widgets_red @rmoff | #ConfluentVUG | @confluentinc

Stream Processing with Kafka Streams Stream: widgets final StreamsBuilder builder = new StreamsBuilder() .stream(“widgets”, Consumed.with(stringSerde, widgetsSerde)) .filter( (key, widget) -> widget.getColour().equals(“RED”) ) .to(“widgets_red”, Produced.with(stringSerde, widgetsSerde)); Stream: widgets_red @rmoff | #ConfluentVUG | @confluentinc

Streams Application Streams Application Streams Application @rmoff | #ConfluentVUG | @confluentinc

Stream Processing with ksqlDB Stream: widgets ksqlDB CREATE STREAM widgets_red AS SELECT * FROM widgets WHERE colour=’RED’; Stream: widgets_red @rmoff | #ConfluentVUG | @confluentinc

} “reading_ts”: “2020-02-14T12:19:27Z”, “sensor_id”: “aa-101”, “production_line”: “w01”, “widget_type”: “acme94”, “temp_celcius”: 23, “widget_weight_g”: 100 Photo by Franck V. on Unsplash { @rmoff | #ConfluentVUG | @confluentinc

SELECT * FROM WIDGETS WHERE WEIGHT_G > 120 { SELECT COUNT(*) FROM WIDGETS GROUP BY PRODUCTION_LINE } “reading_ts”: “2020-02-14T12:19:27Z”, “sensor_id”: “aa-101”, “production_line”: “w01”, “widget_type”: “acme94”, “temp_celcius”: 23, “widget_weight_g”: 100 Photo by Franck V. on Unsplash SELECT AVG(TEMP_CELCIUS) AS TEMP FROM WIDGETS GROUP BY SENSOR_ID HAVING TEMP>20 CREATE SINK CONNECTOR dw WITH ( Object store, ‘connector.class’ = ‘S3Connector’, data warehouse, ‘topics’ = ‘widgets’ RDBMS …); @rmoff | #ConfluentVUG | @confluentinc

Stream Processing with ksqlDB Source stream @rmoff | #ConfluentVUG | @confluentinc

Stream Processing with ksqlDB Source stream Analytics @rmoff | #ConfluentVUG | @confluentinc

Stream Processing with ksqlDB Source stream Applications / Microservices @rmoff | #ConfluentVUG | @confluentinc

Stream Processing with ksqlDB …SUM(TXN_AMT) GROUP BY AC_ID AC _I D= 42 BA LA NC AC E= _I 94 D= .0 42 0 Source stream Applications / Microservices @rmoff | #ConfluentVUG | @confluentinc

ksqlDB or Kafka Streams? @rmoff | #ConfluentVUG | @confluentinc Photo by Ramiz Dedaković on Unsplash

Standing on the Shoulders of Streaming Giants ksqlDB Powered by Ease of use ksqlDB UDFs Kafka Streams Powered by Producer, Consumer APIs Flexibility @rmoff | #ConfluentVUG | @confluentinc