confluentgist

---
version: '2'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.3.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"

confluentgist

import numpy as np
import tensorflow as tf
import tensorflow_io.kafka as kafka_io

# 1. Consume streaming data with Kafka and TensorFlow I/O
def func_x(x):
  # Decode image to (28, 28)
  x = tf.io.decode_raw(x, out_type=tf.uint8)
  x = tf.reshape(x, [28, 28])
  # Convert to float32 for tf.keras

confluentgist

CREATE STREAM products ...;

CREATE STREAM products_repartitioned
   WITH (PARTITIONS=30) AS
   SELECT * FROM products
   EMIT CHANGES;

confluentgist

// Continuously aggregating a KStream into a KTable.
KStream<String, String> locationUpdatesStream = ...;

KTable<String, Long> locationsPerUser
   = locationUpdatesStream
       .groupBy((k, v) -> v.username)
       .count();	

confluentgist

-- Continuously aggregating a stream into a table with a ksqlDB push query.
CREATE STREAM locationUpdatesStream ...;

CREATE TABLE locationsPerUser AS
   SELECT username, COUNT(*)
   FROM locationUpdatesStream
   GROUP BY username
   EMIT CHANGES;

confluentgist

// Create KTable from Kafka topic.
KTable<String, String> table = builder.table("input-topic", Consumed.with(Serdes.String(), Serdes.String()));

confluentgist

-- Create ksqlDB table from Kafka topic.
CREATE TABLE myTable (username VARCHAR, location VARCHAR)
  WITH (KAFKA_TOPIC='input-topic', KEY='username', VALUE_FORMAT='...');

confluentgist

// Create KStream from Kafka topic.
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream =
  builder.stream("input-topic", Consumed.with(Serdes.String(), Serdes.String()));

confluentgist

-- Create ksqlDB stream from Kafka topic.
CREATE STREAM myStream (username VARCHAR, location VARCHAR)
  WITH (KAFKA_TOPIC='input-topic', VALUE_FORMAT='...');

confluentgist

“task”: {
  “type”: “.TaskRole,
  “initialDelayMs”: 20000,
  “taskSpecs”: {
    "bench0": {
      "class": "org.apache.kafka.trogdor.workload.ProduceBenchSpec",
      "startMs": 0,
      "durationMs": 720000,
      "producerNode": "node0",
      "bootstrapServers": "%{bootstrapServers}",