kafka connect & kafka streams/ksql - powerful ecosystem around kafka core

BASEL BERN BRUGG DÜSSELDORF FRANKFURT A.M. FREIBURG I.BR. GENF HAMBURG KOPENHAGEN LAUSANNE MÜNCHEN STUTTGART WIEN ZÜRICH

Kafka Connect & Kafka Streams/KSQLPowerful Ecosystem Around Kafka Core

Guido Schmutz – 5.12.2017

@gschmutz guidoschmutz.wordpress.com

Guido Schmutz

Working at Trivadis for more than 20 yearsOracle ACE Director for Fusion Middleware and SOAConsultant, Trainer Software Architect for Java, Oracle, SOA andBig Data / Fast DataHead of Trivadis Architecture BoardTechnology Manager @ Trivadis

More than 30 years of software development experience

Contact: guido.schmutz@trivadis.comBlog: http://guidoschmutz.wordpress.comSlideshare: http://www.slideshare.net/gschmutzTwitter: gschmutz

Kafka Connect & Kafka Streams/KSQL

Agenda

1. What is Apache Kafka?2. Kafka Connect3. Kafka Streams4. KSQL5. Kafka and "Big Data" / "Fast Data" Ecosystem6. Kafka in Software Architecture

Demo Example

Truck-2 truck/nn/position

Truck-1

Truck-3

mqtt-source

truck_position

detect_dangerous_driving

dangerous_driving

TruckDriver

jdbc-source trucking_driver

join_dangerous_driving_driver

dangerous_driving_driver

consoleconsumer

2016-06-0214:39:56.605|98|27|803014426|Wichita toLittle RockRoute2|Normal|38.65|90.21|5187297736652502631

27,Walter,Ward,Y,24-JUL-85,2017-10-0215:19:00 {"id":27,"firstName":"Walter","lastName":"Ward","available":"Y","birthdate":"24-JUL-85","last_update":1506923052012}

What is Apache Kafka?

Apache Kafka History

2012 2013 2014 2015 2016 2017

Clustermirroringdatacompression

Intra-clusterreplication0.7

DataProcessing(StreamsAPI)

DataIntegration(ConnectAPI)

ExactlyOnceSemanticsPerformanceImprovements

KSQLDeveloperPreview

1.0 JBODSupportSupportJava9

Apache Kafka - Unix Analogy

$ cat < in.txt | grep "kafka" | tr a-z A-Z > out.txt

KafkaConnectAPI KafkaConnectAPIKafkaStreamsAPI

KafkaCore(Cluster)

Adaptedfrom:Confluent

Apache Kafka – A Streaming Platform

High-Level Architecture

Distributed Log at the Core

Scale-Out Architecture

Logs do not (necessarilys) forget

How to get a Kafka environent

On Premises• Bare Metal Installation

• Docker

• Mesos / Kubernetes

• Hadoop Distributions

Cloud• Oracle Event Hub Cloud Service

• Azure HDInsight Kafka

• Confluent Cloud

• …

Demo (I)

Truck-2 truckposition

Truck-1

Truck-3

consoleconsumer

Testdata-GeneratorbyHortonworks

Demo (I) – Create Kafka Topic

$ kafka-topics --zookeeper zookeeper:2181 --create \--topic truck_position --partitions 8 --replication-factor 1

$ kafka-topics --zookeeper zookeeper:2181 –list__consumer_offsets_confluent-metrics_schemasdocker-connect-configsdocker-connect-offsetsdocker-connect-statustruck_position

Demo (I) – Run Producer and Kafka-Console-Consumer

Demo (I) – Java Producer to "truck_position"

Constructing a Kafka Producer

private Properties kafkaProps = new Properties();kafkaProps.put("bootstrap.servers","broker-1:9092);kafkaProps.put("key.serializer", "...StringSerializer");kafkaProps.put("value.serializer", "...StringSerializer");

producer = new KafkaProducer<String, String>(kafkaProps);

ProducerRecord<String, String> record =new ProducerRecord<>("truck_position", driverId, eventData);

try {metadata = producer.send(record).get();

} catch (Exception e) {}

Demo (II) – devices send to MQTT instead of Kafka

Truck-1

Truck-3

Demo (II) – devices send to MQTT instead of Kafka

Demo (II) - devices send to MQTT instead of Kafka –how to get the data into Kafka?

Truck-1

Truck-3

truckposition raw

Kafka Connect

Kafka Connect - Overview

SourceConnector

SinkConnector

Kafka Connect – Single Message Transforms (SMT)

Simple Transformations for a single message

Defined as part of Kafka Connect• some useful transforms provided out-of-the-box• Easily implement your own

Optionally deploy 1+ transforms with each connector• Modify messages produced by source

connector• Modify messages sent to sink connectors

Makes it much easier to mix and match connectors

Some of currently available transforms:• InsertField• ReplaceField• MaskField• ValueToKey• ExtractField• TimestampRouter• RegexRouter• SetSchemaMetaData• Flatten• TimestampConverter

Kafka Connect – Many Connectors

60+ since first release (0.9+)

20+ from Confluent and Partners

Source:http://www.confluent.io/product/connectors

ConfluentsupportedConnectors

CertifiedConnectors CommunityConnectors

Demo (III)

Truck-1

Truck-3

mqtt tokafka

truck_position

consoleconsumer

Demo (III) – Create MQTT Connect through REST API #!/bin/bashcurl -X "POST" "http://192.168.69.138:8083/connectors" \

-H "Content-Type: application/json" \-d $'{

"name": "mqtt-source","config": {"connector.class":

"com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector","connect.mqtt.connection.timeout": "1000","tasks.max": "1","connect.mqtt.kcql":

"INSERT INTO truck_position SELECT * FROM truck/+/position","name": "MqttSourceConnector","connect.mqtt.service.quality": "0", "connect.mqtt.client.id": "tm-mqtt-connect-01","connect.mqtt.converter.throw.on.error": "true","connect.mqtt.hosts": "tcp://mosquitto:1883"}

Demo (III) – Call REST API and Kafka Console Consumer

Demo (III)

Truck-1

Truck-3

mqtt tokafka

truck_position

consoleconsumer

whataboutsomeanalytics?

Kafka Streams

Kafka Streams - Overview

• Designed as a simple and lightweight library in Apache Kafka

• no external dependencies on systems other than Apache Kafka

• Part of open source Apache Kafka, introduced in 0.10+• Leverages Kafka as its internal messaging layer• Supports fault-tolerant local state• Event-at-a-time processing (not microbatch) with millisecond

latency• Windowing with out-of-order data using a Google DataFlow-like

Kafka Stream DSL and Processor Topology

KStream<Integer, String> stream1 =builder.stream("in-1");

KStream<Integer, String> stream2=builder.stream("in-2");

KStream<Integer, String> joined =stream1.leftJoin(stream2, …);

KTable<> aggregated = joined.groupBy(…).count("store");

aggregated.to("out-1");

Kafka Stream DSL and Processor Topology

KStream<Integer, String> stream1 =builder.stream("in-1");

KStream<Integer, String> stream2=builder.stream("in-2");

KStream<Integer, String> joined =stream1.leftJoin(stream2, …);

KTable<> aggregated = joined.groupBy(…).count("store");

aggregated.to("out-1");

Kafka Streams Cluster

Processor Topology

Kafka Cluster

input-1

input-2

store(changelog)

output

tState

Kafka Cluster

Processor Topology

input-1Partition0

Partition1

Partition2

Partition3

input-2Partition0

Partition1

Partition2

Partition3

Kafka Streams 1

Kafka Streams 2

Kafka Cluster

Processor Topology

input-1Partition0

Partition1

Partition2

Partition3

input-2Partition0

Partition1

Partition2

Partition3

Kafka Streams 1 Kafka Streams 2

Kafka Streams 3 Kafka Streams 4

Stream vs. Table

Event Stream State Stream (Change Log Stream)

2017-10-02T20:18:46 11,Normal,41.87,-87.67

2017-10-02T20:18:55 11,Normal,40.38,-89.17

2017-10-02T20:18:59 21,Normal,42.23,-91.78

2017-10-02T20:19:01 21,Normal,41.71,-91.32

2017-10-02T20:19:02 11,Normal,38.65,-90.2

2017-10-02T20:19:23 21,Normal41.71,-91.32

11 2017-10-02T20:18:46,11,Normal,41.87,-87.67

11 2017-10-02T20:18:55,11,Normal,40.38,-89.17

21 2017-10-02T20:18:59,21,Normal,42.23,-91.78

21 2017-10-02T20:19:01,21,Normal,41.71,-91.32

11 2017-10-02T20:19:02,11,Normal,38.65,-90.2

21 2017-10-02T20:19:23,21,Normal41.71,-91.32

KStream KTable

Kafka Streams: Key Features

• Native, 100%-compatible Kafka integration• Secure stream processing using Kafka's security features• Elastic and highly scalable• Fault-tolerant• Stateful and stateless computations• Interactive queries• Time model• Windowing• Supports late-arriving and out-of-order data• Millisecond processing latency, no micro-batching• At-least-once and exactly-once processing guarantees

Demo (IV)

Truck-1

Truck-3

mqtt tokafka

truck_position_s

dangerous_driving

consoleconsumer

Demo (IV) - Create Stream

final KStreamBuilder builder = new KStreamBuilder();

KStream<String, String> source = builder.stream(stringSerde, stringSerde, "truck_position");

KStream<String, TruckPosition> positions = source.map((key,value) ->

new KeyValue<>(key, TruckPosition.create(value)));

KStream<String, TruckPosition> filtered = positions.filter(TruckPosition::filterNonNORMAL);

filtered.map((key,value) -> new KeyValue<>(key,value._originalRecord))

.to("dangerous_driving");

KSQL: a Streaming SQL Engine for Apache Kafka

• Enables stream processing with zero coding required• The simples way to process streams of data in real-time• Powered by Kafka and Kafka Streams: scalable, distributed, mature• All you need is Kafka – no complex deployments• available as Developer preview!

• STREAM and TABLE as first-class citizens• STREAM = data in motion• TABLE = collected state of a stream• join STREAM and TABLE

KSQL Deployment Models

Standalone Mode Cluster Mode

Source:Confluent

Demo (IV)

Truck-1

Truck-3

mqtt tokafka

truck_position_s

dangerous_driving

consoleconsumer

Demo (V) - Start Kafka KSQL$ docker-compose exec ksql-cli ksql-cli local --bootstrap-server broker-1:9092

======================================= _ __ _____ ____ _ == | |/ // ____|/ __ \| | == | ' /| (___ | | | | | == | < \___ \| | | | | == | . \ ____) | |__| | |____ == |_|\_\_____/ \___\_\______| == == Streaming SQL Engine for Kafka =

CLI v0.1, Server v0.1 located at http://localhost:9098

Having trouble? Type 'help' (case-insensitive) for a rundown of how things work!

Demo (IV) - Create Streamksql> CREATE STREAM truck_position_s \(ts VARCHAR, \truckid VARCHAR, \driverid BIGINT, \routeid BIGINT, \routename VARCHAR, \eventtype VARCHAR, \latitude DOUBLE, \longitude DOUBLE, \correlationid VARCHAR) \WITH (kafka_topic='truck_position', \

value_format='DELIMITED');

Message----------------Stream created

ksql> describe truck_position_s;

ksql> SELECT * FROM truck_position_s;

1506922133306 | "truck/13/position0 | �2017-10-02T07:28:53 | 31 | 13 | 371182829 | Memphis to Little Rock | Normal | 41.76 | -89.6 | -20842639519146641061506922133396 | "truck/16/position0 | �2017-10-02T07:28:53 | 19 | 16 | 160405074 | Joplin to Kansas City Route 2 | Normal | 41.48 | -88.07 | -20842639519146641061506922133457 | "truck/30/position0 | �2017-10-02T07:28:53 | 26 | 30 | 160779139 | Des Moines to Chicago Route 2 | Normal | 41.85 | -89.29 | -20842639519146641061506922133485 | "truck/23/position0 | �2017-10-02T07:28:53 | 32 | 23 | 1090292248 | Peoria to Ceder Rapids Route 2 | Normal | 41.48 | -88.07 | -20842639519146641061506922133497 | "truck/12/position0 | �2017-10-02T07:28:53 | 80 | 12 | 1961634315 | Saint Louis to Memphis | Normal | 41.74 | -91.47 | -20842639519146641061506922133547 | "truck/14/position0 | �2017-10-02T07:28:53 | 73 | 14 | 1927624662 | Springfield to KC Via Columbia | Normal | 35.12 | -90.68 | -2084263951914664106

ksql> SELECT * FROM truck_position_s WHERE eventtype != 'Normal';

1506922264016 | "truck/11/position0 | �2017-10-02T07:31:04 | 27 | 11 | 1325712174 | Saint Louis to Tulsa Route2 | Lane Departure | 38.5 | -90.69 | -20842639519146641061506922281156 | "truck/11/position0 | �2017-10-02T07:31:21 | 27 | 11 | 1325712174 | Saint Louis to Tulsa Route2 | Unsafe tail distance | 37.81 | -92.31 | -20842639519146641061506922284436 | "truck/10/position0 | �2017-10-02T07:31:24 | 93 | 10 | 1384345811 | Joplin to Kansas City | Unsafe following distance | 37.02 | -94.54 | -20842639519146641061506922297887 | "truck/11/position0 | �2017-10-02T07:31:37 | 27 | 11 | 1325712174 | Saint Louis to Tulsa Route2 | Unsafe following distance | 37.09 | -94.23 | -2084263951914664106

ksql> CREATE STREAM dangerous_driving_s \WITH (kafka_topic= dangerous_driving', \

value_format='DELIMITED') \AS SELECT * FROM truck_position_s \WHERE eventtype != 'Normal';

Message----------------------------Stream created and running

ksql> select * from dangerous_driving_s;1506922849375 | "truck/11/position0 | �2017-10-02T07:40:49 | 90 | 11 | 160779139 | Des Moines to Chicago Route 2 | Overspeed | 41.48 | -88.07 | 35691830713478983661506922866488 | "truck/11/position0 | �2017-10-02T07:41:06 | 90 | 11 | 160779139 | Des Moines to Chicago Route 2 | Overspeed | 40.38 | -89.17 | 3569183071347898366

ksql> describe dangerous_driving_s;

Demo (V)

Truck-1

Truck-3

mqtt-source

truck_position

dangerous_driving

TruckDriver

jdbc-source trucking_driver

join_dangerous_driving_driver

dangerous_driving_driver

27,Walter,Ward,Y,24-JUL-85,2017-10-0215:19:00

consoleconsumer

{"id":27,"firstName":"Walter","lastName":"Ward","available":"Y","birthdate":"24-JUL-85","last_update":1506923052012}

Demo (V) – Create JDBC Connect through REST API #!/bin/bashcurl -X "POST" "http://192.168.69.138:8083/connectors" \

-H "Content-Type: application/json" \-d $'{

"name": "jdbc-driver-source","config": {

"connector.class": "JdbcSourceConnector","connection.url":"jdbc:postgresql://db/sample?user=sample&password=sample","mode": "timestamp","timestamp.column.name":"last_update","table.whitelist":"driver","validate.non.null":"false","topic.prefix":"trucking_","key.converter":"org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable": "false","value.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter.schemas.enable": "false","name": "jdbc-driver-source","transforms":"createKey,extractInt", "transforms.createKey.type":"org.apache.kafka.connect.transforms.ValueToKey", "transforms.createKey.fields":"id", "transforms.extractInt.type":"org.apache.kafka.connect.transforms.ExtractField$Key", "transforms.extractInt.field":"id"

Demo (V) – Create JDBC Connect through REST API

Demo (V) - Create Table with Driver Stateksql> CREATE TABLE driver_t \

(id BIGINT, \first_name VARCHAR, \last_name VARCHAR, \available VARCHAR) \WITH (kafka_topic='trucking_driver', \

value_format='JSON');Message----------------Table created

Demo (V) - Create Table with Driver Stateksql> CREATE STREAM dangerous_driving_and_driver_s \WITH (kafka_topic='dangerous_driving_and_driver_s', \

value_format='JSON') \AS SELECT driverid, first_name, last_name, truckid, routeid,routename, eventtype \FROM truck_position_s \LEFT JOIN driver_t \ON dangerous_driving_and_driver_s.driverid = driver_t.id;

Message----------------------------Stream created and running

ksql> select * from dangerous_driving_and_driver_s;1511173352906 | 21 | 21 | Lila | Page | 58 | 1594289134 | Memphis to Little Rock Route 2 | Unsafe tail distance1511173353669 | 12 | 12 | Laurence | Lindsey | 93 | 1384345811 | Joplin to KansasCity | Lane Departure1511173435385 | 11 | 11 | Micky | Isaacson | 22 | 1198242881 | Saint Louis to Chicago Route2 | Unsafe tail distance

Kafka and "Big Data" / "Fast Data"Ecosystem

Kafka and the Big Data / Fast Data ecosystem

Kafka integrates with many popular products / frameworks

• Apache Spark Streaming

• Apache Flink

• Apache Storm

• Apache Apex

• Apache NiFi

• StreamSets

• Oracle Stream Analytics

• Oracle Service Bus

• Oracle GoldenGate

• Oracle Event Hub Cloud Service

• Debezium CDC

• …

AdditionalInfo:https://cwiki.apache.org/confluence/display/KAFKA/EcosystemKafka Connect & Kafka Streams/KSQL

Kafka in Software Architecture

Hadoop ClusterdHadoop Cluster

Big Data Cluster

Traditional Big Data Architecture

BITools

Enterprise Data Warehouse

Billing &Ordering

CRM / Profile

MarketingCampaigns

File Import / SQL Import

Search/Explore

Online&MobileApps

Search

Parallel BatchProcessing

DistributedFilesystem

• MachineLearning• GraphAlgorithms• NaturalLanguageProcessing

Event HubEvent

Hadoop ClusterdHadoop Cluster

Big Data Cluster

Event Hub – handle event stream data

BITools

Location

Social

Clickstream

Sensor Data

Billing &Ordering

CRM / Profile

MarketingCampaigns

Event Hub

CallCenter

WeatherData

MobileApps

Search/Explore

Online&MobileApps

Search

Data Flow

• MachineLearning• GraphAlgorithms• NaturalLanguageProcessing

Hadoop ClusterdHadoop ClusterBig Data Cluster

Event Hub – taking Velocity into account

Location

Social

Clickstream

Sensor Data

Billing &Ordering

CRM / Profile

MarketingCampaigns

CallCenter

MobileApps

Batch Analytics

Streaming Analytics

Results

Stream AnalyticsNoSQL

Reference /Models

Search

Dashboard

BITools

Search/Explore

Online&MobileApps

WeatherData

Event HubEvent

HubEvent Hub

Container

Hadoop ClusterdHadoop ClusterBig Data Cluster

Event Hub – Asynchronous Microservice Architecture

Location

Social

Clickstream

Sensor Data

Billing &Ordering

CRM / Profile

MarketingCampaigns

CallCenter

MobileApps

ParallelBatch

ProcessingDistributedFilesystem

Microservice

NoSQLRDBMS

Search

BITools

Search/Explore

Online&MobileApps

WeatherData

Event HubEvent

HubEvent Hub

Technology on its own won't help you.You need to know how to use it properly.

kafka connect & kafka streams/ksql - powerful ecosystem around kafka core

Technology

kafka streams - power without weight (jeeconf)

kafka streams: the stream processing engine of apache kafka

the rise of the streaming platform - big data paris...

back-pressure in action: handling high-burst workloads with...

kafka streams + deep learning - rainfocus streams + deep...

introducing apache kafka's streams api - kafka meetup...

architecting fast data applications - techarch day · kafka...

ibm event streams using apache kafka · apache kafka...

a system to lower barriers to support the billion object...

scylla summit 2017: streaming etl in kafka for everyone with...

machine learning and deep learning applied to real time with...

kafka streams: hands-on session - ce.uniroma2.it · kafka...

event stream processing using kafka...

architecture guide real-time data streaming · the kafka...

real time streams at scale with kafka: couchbase connect...

ksql intro

introduction to kafka streams - michael noll - berlin...

data stream processing - amazon s3 · • high performance...

introducing kafka streams: large-scale stream processing...

back-pressure in action: handling high-burst workloads with...