Working with KSQL Streams

Start a KSQL session:

sudo ksql

Set auto.offset.reset to earliest:

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

Look at the data in the member_signups topic:

PRINT 'member_signups' FROM BEGINNING;

Create a stream from the topic:

CREATE STREAM member_signups
  (firstname VARCHAR,
    lastname VARCHAR,
    email_notifications BOOLEAN)
  WITH (KAFKA_TOPIC='member_signups',
    VALUE_FORMAT='DELIMITED');

Create the persistent streaming query:

CREATE STREAM member_signups_email AS
  SELECT * FROM member_signups WHERE email_notifications=true;

View the data in the output topic to verify that everything is working:

PRINT 'MEMBER_SIGNUPS_EMAIL' FROM BEGINNING;