Upgrade to Pro — share decks privately, control downloads, hide ads and more …

What's this Stream Processing stuff anyway?

What's this Stream Processing stuff anyway?

Oak Table World 2017 - talk from Gwen Shapira and Robin Moffatt, all about Apache Kafka, Kafka Connect, and KSQL

Robin Moffatt

October 03, 2017
Tweet

More Decks by Robin Moffatt

Other Decks in Technology

Transcript

  1. 1 What's this Stream Processing stuff anyway? Oak Table World

    2017 Gwen Shapira & Robin Moffatt Confluent @rmoff [email protected] @gwenshap [email protected]
  2. 2 Let’s take a trip back in time. Each application

    has its own database for storing information. But we want that information elsewhere for analytics and reporting.
  3. 3 We don't want to query the transactional system, so

    we create a process to extract from the source to a data warehouse / lake
  4. 4 Let’s take a trip back in time We want

    to unify data from multiple systems, so create conformed dimensions and batch processes to federate our data. This is all batch driven, so latency is built in by design.
  5. 5 Let’s take a trip back in time As well

    as our data warehouse, we want to use our transactional data to populate search replicas, Graph databases, noSQL stores…all introducing more point-to-point dependencies in our system
  6. 6 Let’s take a trip back in time Ultimately we

    end up with a spaghetti architecture. It can't scale easily, it's tightly coupled, it's generally batch-driven and we can't get data when we want it where we want it.
  7. 8 Apache Kafka, a distributed streaming platform, enables us to

    decouple all our applications creating data from those utilising it. We can create low- latency streams of data, transformed as necessary.
  8. 10 Happy days! We can actually build streaming data pipelines

    using just our bare hands, configuration files, and SQL.
  9. 12 $ cat speakers.txt • Gwen Shapira • Product Manager

    & Kafka Committer • @gwenshap • Robin Moffatt • Partner Technology Evangelist @ Confluent • @rmoff
  10. 13

  11. 14

  12. 15

  13. 16

  14. 18 Streaming Application Data to Kafka • Applications are rich

    source of events • Modifying applications is not always possible or desirable • And what if the data gets changed within the database or by other apps? • JDBC is one option for extracting data • Confluent Open Source includes JDBC source & sink connectors
  15. 19 Liberate Application Data into Kafka with CDC • Relational

    databases use transaction logs to ensure Durability of data • Change-Data-Capture (CDC) mines the log to get raw events from the database • CDC tools that integrate with Kafka Connect include: • Debezium • DBVisit • GoldenGate • Attunity • + more
  16. 20 Single Message Transform (SMT) -- Extract, TRANSFORM, Load… •

    Modify events before storing in Kafka: • Mask/drop sensitive information • Set partitioning key • Store lineage • Modify events going out of Kafka: • Route high priority events to faster data stores • Direct events to different Elasticsearch indexes • Cast data types to match destination
  17. 22 KSQL from Confluent A Developer Preview of KSQL An

    Open Source Streaming SQL Engine for Apache KafkaTM
  18. 23 KSQL: a Streaming SQL Engine for Apache Kafka™ from

    Confluent • Enables stream processing with zero coding required • The simplest way to process streams of data in real-time • Powered by Kafka: scalable, distributed, battle-tested • All you need is Kafka–No complex deployments of bespoke systems for stream processing Ksql>
  19. 24 CREATE STREAM possible_fraud AS SELECT card_number, count(*) FROM authorization_attempts

    WINDOW TUMBLING (SIZE 5 SECONDS) GROUP BY card_number HAVING count(*) > 3; KSQL: the Simplest Way to Do Stream Processing
  20. 25 KSQL Concepts • STREAM and TABLE as first-class citizens

    • Interpretations of topic content • STREAM - data in motion • TABLE - collected state of a stream • One record per key (per window) • Current values (compacted topic) • STREAM – TABLE Joins
  21. 26 Window Aggregations Three types supported (same as KStreams): •

    TUMBLING: Fixed-size, non-overlapping, gap-less windows • SELECT ip, count(*) AS hits FROM clickstream WINDOW TUMBLING (size 1 minute) GROUP BY ip; • HOPPING: Fixed-size, overlapping windows • SELECT ip, SUM(bytes) AS bytes_per_ip_and_bucket FROM clickstream WINDOW HOPPING ( size 20 second, advance by 5 second) GROUP BY ip; • SESSION: Dynamically-sized, non-overlapping, data-driven window • SELECT ip, SUM(bytes) AS bytes_per_ip FROM clickstream WINDOW SESSION (20 second) GROUP BY ip; More: http://docs.confluent.io/current/streams/developer-guide.html#windowing
  22. 39 KSQL in action ksql> CREATE stream rental (rental_id INT,

    rental_date INT, inventory_id INT, customer_id INT, return_date INT, staff_id INT, last_update INT ) WITH (kafka_topic = 'sakila-rental', value_format = 'json'); Message ---------------- Stream created * Command formatted for clarity here. Linebreaks need to be denoted by \ in KSQL
  23. 40 KSQL in action ksql> describe rental; Field | Type

    -------------------------------- ROWTIME | BIGINT ROWKEY | VARCHAR(STRING) RENTAL_ID | INTEGER RENTAL_DATE | INTEGER INVENTORY_ID | INTEGER CUSTOMER_ID | INTEGER RETURN_DATE | INTEGER STAFF_ID | INTEGER LAST_UPDATE | INTEGER
  24. 41 KSQL in action ksql> select * from rental limit

    3; 1505830937567 | null | 1 | 280113040 | 367 | 130 | 1505830937567 | null | 2 | 280176040 | 1525 | 459 | 1505830937569 | null | 3 | 280722040 | 1711 | 408 |
  25. 42 KSQL in action SELECT rental_id , TIMESTAMPTOSTRING(rental_date, 'yyyy-MM-dd HH:mm:ss.SSS'),

    TIMESTAMPTOSTRING(return_date, 'yyyy-MM-dd HH:mm:ss.SSS') FROM rental limit 3; 1 | 2005-05-24 22:53:30.000 | 2005-05-26 22:04:30.000 2 | 2005-05-24 22:54:33.000 | 2005-05-28 19:40:33.000 3 | 2005-05-24 23:03:39.000 | 2005-06-01 22:12:39.000 LIMIT reached for the partition. Query terminated ksql>
  26. 43 KSQL in action SELECT rental_id , TIMESTAMPTOSTRING(rental_date, 'yyyy-MM-dd HH:mm:ss.SSS'),

    TIMESTAMPTOSTRING(return_date, 'yyyy-MM-dd HH:mm:ss.SSS'), ceil((cast(return_date AS DOUBLE) – cast(rental_date AS DOUBLE) ) / 60 / 60 / 24 / 1000) FROM rental; 1 | 2005-05-24 22:53:30.000 | 2005-05-26 22:04:30.000 | 2.0 2 | 2005-05-24 22:54:33.000 | 2005-05-28 19:40:33.000 | 4.0 3 | 2005-05-24 23:03:39.000 | 2005-06-01 22:12:39.000 | 8.0
  27. 44 KSQL in action CREATE stream rental_lengths AS SELECT rental_id

    , TIMESTAMPTOSTRING(rental_date, 'yyyy-MM-dd HH:mm:ss.SSS') , TIMESTAMPTOSTRING(return_date, 'yyyy-MM-dd HH:mm:ss.SSS') , ceil(( cast(return_date AS DOUBLE) – cast( rental_date AS DOUBLE) ) / 60 / 60 / 24 / 1000) FROM rental;
  28. 45 KSQL in action ksql> select rental_id, rental_date, return_date, RENTAL_LENGTH_DAYS

    from rental_lengths; 3 | 2005-05-24 23:03:39.000 | 2005-06-01 22:12:39.000 | 8.0 4 | 2005-05-24 23:04:41.000 | 2005-06-03 01:43:41.000 | 10.0 7 | 2005-05-24 23:11:53.000 | 2005-05-29 20:34:53.000 | 5.0
  29. 46 KSQL in action $ kafka-topics --zookeeper localhost:2181 --list RENTAL_LENGTHS

    $ kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic RENTAL_LENGTHS | jq '.' { "RENTAL_DATE": "2005-05-24 22:53:30.000", "RENTAL_LENGTH_DAYS": 2, "RETURN_DATE": "2005-05-26 22:04:30.000", "RENTAL_ID": 1 }
  30. 47 KSQL in action CREATE stream long_rentals AS SELECT *

    FROM rental_lengths WHERE rental_length_days > 7; ksql> select rental_id, rental_date, return_date, RENTAL_LENGTH_DAYS from long_rentals; 3 | 2005-05-24 23:03:39.000 | 2005-06-01 22:12:39.000 | 8.0 4 | 2005-05-24 23:04:41.000 | 2005-06-03 01:43:41.000 | 10.0
  31. 48 KSQL in action $ kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic

    LONG_RENTALS | jq '.' { "RENTAL_DATE": " 2005-05-24 23:03:39.000", "RENTAL_LENGTH_DAYS": 8, "RETURN_DATE": " 2005-06-01 22:12:39.000", "RENTAL_ID": 3 }
  32. 49 Streaming ETL with Kafka Connect and KSQL MySQL Kafka

    Connect Kafka Cluster rental rental_lengths long_rentals Elasticsearch CREATE STREAM RENTAL_LENGTHS AS SELECT END_DATE - START_DATE […] FROM RENTAL Kafka Connect CREATE STREAM LONG_RENTALS AS SELECT … FROM RENTAL_LENGTHS WHERE DURATION > 14
  33. 52 Kafka Connect to stream Kafka Topics to Elasticsearch…MySQL…& more

    { "name": "es-sink-avro-02", "config": { "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "connection.url": "http://localhost:9200", "type.name": "type.name=kafka-connect", "topics": "sakila-avro-rental", "key.ignore": "true", "transforms":"dropPrefix", "transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter", "transforms.dropPrefix.regex":"sakila-avro-(.*)", "transforms.dropPrefix.replacement":"$1" } }
  34. 55 Kafka Connect + Schema Registry = WIN MySQL Avro

    Message Elasticsearch Schema Registry Avro Schema Kafka Connect Kafka Connect
  35. 56 Kafka Connect + Schema Registry = WIN MySQL Avro

    Message Elasticsearch Schema Registry Avro Schema Kafka Connect Kafka Connect
  36. 59 Kafka Connect to stream Kafka Topics to Elasticsearch…MySQL…& more

    { "name": "es-sink-rental-lengths-02", "config": { "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "false", "value.converter.schemas.enable": "false", "schema.ignore": "true", "connection.url": "http://localhost:9200", "type.name": "type.name=kafka-connect", "topics": "RENTAL_LENGTHS", "topic.index.map": "RENTAL_LENGTHS:rental_lengths", "key.ignore": "true" } }
  37. 62 Streaming ETL with Apache Kafka and Confluent Platform –

    no coding! MySQL Elasticsearch Kafka Connect Kafka Connect Kafka Cluster KSQL Kafka Streams
  38. 63

  39. 65 Confluent Platform: Enterprise Streaming based on Apache Kafka™ Database

    Changes Log Events loT Data Web Events … CRM Data Warehouse Database Hadoop Data Integration … Monitoring Analytics Custom Apps Transformations Real-time Applications … Apache Open Source Confluent Open Source Confluent Enterprise Confluent Platform Confluent Platform Apache Kafka™ Core | Connect API | Streams API Data Compatibility Schema Registry Monitoring & Administration Confluent Control Center | Security Operations Replicator | Auto Data Balancing Development and Connectivity Clients | Connectors | REST Proxy | KSQL | CLI
  40. 66