Confluent Open Source Quick Start (Docker)

This quick start demonstrates how to get up and running with Confluent Open Source and its main components using Docker containers. It demonstrates the basic and most powerful capabilities, including creating topics, adding and modifying data, and stream processing by using KSQL. In this quick start you will create Kafka topics and streaming queries on these topics by using KSQL.

This quick start leverages the Confluent Platform CLI, the Kafka CLI, and the KSQL CLI. For a rich UI-based experience, try out the Confluent Enterprise quick start.

Prerequisites
  • Docker
    • Docker version 1.11 or later is installed and running.
    • Docker Compose is installed. It is installed by default with Docker for Mac and Windows.
    • Docker memory resource is allocated minimally at 8 MB.
  • Git

Step 1: Download and Start Confluent Platform Using Docker

  1. Clone the Confluent Platform Docker Images GitHub Repository and checkout the latest branch.

    git clone https://github.com/confluentinc/cp-docker-images
    
  2. Navigate to examples directory (/cp-docker-images/examples/cp-all-in-one/).

  3. Start Confluent Platform in detached mode (-d).

    docker-compose up -d
    

    This starts the Confluent Platform with separate containers for all Confluent Platform components. Your output should resemble the following:

    Creating network "cp-all-in-one_default" with the default driver
    Creating zookeeper ... done
    Creating broker    ... done
    Creating schema-registry ... done
    Creating rest-proxy      ... done
    Creating connect         ... done
    Creating ksql-datagen    ... done
    Creating ksql-server     ... done
    Creating control-center  ... done
    Creating ksql-cli        ... done
    
  4. Optional: Run this command to verify that the services are up and running:

    docker-compose ps
    

    You should see the following:

         Name                    Command               State                Ports
    ------------------------------------------------------------------------------------------
    broker            /etc/confluent/docker/run        Up      0.0.0.0:29092->29092/tcp,
                                                               0.0.0.0:9092->9092/tcp
    connect           /etc/confluent/docker/run        Up      0.0.0.0:8083->8083/tcp,
                                                               9092/tcp
    control-center    /etc/confluent/docker/run        Up      0.0.0.0:9021->9021/tcp
    ksql-cli          ksql http://localhost:8088       Up
    ksql-datagen      bash -c echo Waiting for K ...   Up
    ksql-server       /etc/confluent/docker/run        Up      0.0.0.0:8088->8088/tcp
    rest-proxy        /etc/confluent/docker/run        Up      0.0.0.0:8082->8082/tcp
    schema-registry   /etc/confluent/docker/run        Up      0.0.0.0:8081->8081/tcp
    zookeeper         /etc/confluent/docker/run        Up      0.0.0.0:2181->2181/tcp,
                                                               2888/tcp, 3888/tcp
    

    If the state is not Up, rerun the docker-compose up -d command.

Step 2: Create Kafka Topics

In this step Kafka topics are created in Confluent Platform by using the Kafka CLI.

  1. Run this command to create a topic named users.

    docker-compose exec broker kafka-topics --create --zookeeper \
    zookeeper:2181 --replication-factor 1 --partitions 1 --topic users
    

    Your output should resemble:

    Created topic "users".
    
  2. Run this command to create a topic named pageviews.

    docker-compose exec broker kafka-topics --create --zookeeper \
    zookeeper:2181 --replication-factor 1 --partitions 1 --topic pageviews
    

    Your output should resemble:

    Created topic "pageviews".
    

Step 3: Create Sample Data

In this step you create sample data for the Kafka topics pageviews and users by using the ksql-datagen.

Tip

The datagen runs as a long-running process in your terminal. Run each datagen step in a separate terminal.

  1. Produce Kafka data to the pageviews topic using the data generator. The following example continuously generates data with a value in DELIMITED format.

    docker-compose exec ksql-datagen ksql-datagen quickstart=pageviews format=delimited topic=pageviews maxInterval=100 \
    bootstrap-server=broker:9092
    
  2. Produce Kafka data to the users topic using the data generator. The following example continuously generates data with a value in JSON format.

    docker-compose exec ksql-datagen ksql-datagen quickstart=users format=json topic=users maxInterval=1000 \
    bootstrap-server=broker:9092
    

Step 4: Create and Write to a Stream and Table using KSQL

In this step KSQL queries are run on the pageviews and users topics that were created in the previous step. The following KSQL commands are run from the KSQL CLI. Enter these commands in your terminal and press Enter.

Important

  • Confluent Platform must be installed and running.
  • To try out the preview KSQL web interface, see the Confluent Enterprise quick start.
  • All KSQL commands must end with a closing semicolon (;).

Create Streams and Tables

  1. Start the KSQL CLI in your terminal with this command.

    $ LOG_DIR=./ksql_logs <path-to-confluent>/bin/ksql
    
    1. Optional: Examine query run-time metrics and details. Observe that information including the target Kafka topic is available, as well as throughput figures for the messages being processed.
  2. Create a stream (pageviews) from the Kafka topic pageviews, specifying the value_format of DELIMITED.

    ksql> CREATE STREAM pageviews (viewtime BIGINT, userid VARCHAR, pageid VARCHAR) \
    WITH (KAFKA_TOPIC='pageviews', VALUE_FORMAT='DELIMITED');
    

    Tip: Enter the SHOW STREAMS; command to view your streams. For example:

     Stream Name      | Kafka Topic      | Format
    -------------------------------------------------
     PAGEVIEWS        | pageviews        | DELIMITED
    -------------------------------------------------
    
  3. Create a table (users) with several columns from the Kafka topic users, with the value_format of JSON.

    ksql> CREATE TABLE users (registertime BIGINT, gender VARCHAR, regionid VARCHAR, \
    userid VARCHAR, \interests array<VARCHAR>, contact_info map<VARCHAR, VARCHAR>) \
    WITH (KAFKA_TOPIC='users', VALUE_FORMAT='JSON', KEY = 'userid');
    

    Tip: Enter the SHOW TABLES; query to view your tables.

     Table Name        | Kafka Topic       | Format    | Windowed
    --------------------------------------------------------------
     USERS             | users             | JSON      | false
    --------------------------------------------------------------
    

Write Queries

These examples write queries using KSQL. The following KSQL commands are run from the KSQL CLI. Enter these commands in your terminal and press Enter.

  1. Add the custom query property earliest for the auto.offset.reset parameter. This instructs KSQL queries to read all available topic data from the beginning. This configuration is used for each subsequent query. For more information, see the KSQL Configuration Parameter Reference.

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

    Your output should resemble:

    Successfully changed local property 'auto.offset.reset' from 'null' to 'earliest'
    
  2. Create a query that returns data from a stream with the results limited to three rows.

    ksql> SELECT pageid FROM pageviews LIMIT 3;
    

    Your output should resemble:

    Page_45
    Page_38
    Page_11
    LIMIT reached for the partition.
    Query terminated
    
  3. Create a persistent query that filters for female users. The results from this query are written to the Kafka PAGEVIEWS_FEMALE topic. This query enriches the pageviews STREAM by doing a LEFT JOIN with the users TABLE on the user ID, where a condition (gender = 'FEMALE') is met.

    ksql> CREATE STREAM pageviews_female AS SELECT users.userid AS userid, pageid, \
    regionid, gender FROM pageviews LEFT JOIN users ON pageviews.userid = users.userid \
    WHERE gender = 'FEMALE';
    

    Your output should resemble:

     Message
    ----------------------------
     Stream created and running
    ----------------------------
    
  4. Create a persistent query where a condition (regionid) is met, using LIKE. Results from this query are written to a Kafka topic named pageviews_enriched_r8_r9.

    ksql> CREATE STREAM pageviews_female_like_89 WITH (kafka_topic='pageviews_enriched_r8_r9', \
    value_format='DELIMITED') AS SELECT * FROM pageviews_female WHERE regionid LIKE '%_8' OR regionid LIKE '%_9';
    

    Your output should resemble:

     Message
    ----------------------------
     Stream created and running
    ----------------------------
    
  5. Create a persistent query that counts the pageviews for each region and gender combination in a tumbling window of 30 seconds when the count is greater than 1. Because the procedure is grouping and counting, the result is now a table, rather than a stream. Results from this query are written to a Kafka topic called PAGEVIEWS_REGIONS.

    ksql> CREATE TABLE pageviews_regions AS SELECT gender, regionid , \
    COUNT(*) AS numusers FROM pageviews_female WINDOW TUMBLING (size 30 second) \
    GROUP BY gender, regionid HAVING COUNT(*) > 1;
    

    Your output should resemble:

     Message
    ---------------------------
     Table created and running
    ---------------------------
    

Step 5: Monitor Streaming Data

Now that your streams are running you can monitor them.

  • View the details for your stream or table with the DESCRIBE EXTENDED command. For example, run this command to view the pageviews_female_like_89 stream:

    DESCRIBE EXTENDED pageviews_female_like_89;
    

    Your output should look like this:

    Type                 : STREAM
    Key field            : PAGEVIEWS.USERID
    Timestamp field      : Not set - using <ROWTIME>
    Key format           : STRING
    Value format         : DELIMITED
    Kafka output topic   : pageviews_enriched_r8_r9 (partitions: 4, replication: 1)
    
     Field    | Type
    --------------------------------------
     ROWTIME  | BIGINT           (system)
     ROWKEY   | VARCHAR(STRING)  (system)
     USERID   | VARCHAR(STRING)  (key)
     PAGEID   | VARCHAR(STRING)
     REGIONID | VARCHAR(STRING)
     GENDER   | VARCHAR(STRING)
    --------------------------------------
    
    Queries that write into this STREAM
    -----------------------------------
    id:CSAS_PAGEVIEWS_FEMALE_LIKE_89 - CREATE STREAM pageviews_female_like_89 WITH (kafka_topic='pageviews_enriched_r8_r9', value_format='DELIMITED') AS SELECT * FROM pageviews_female WHERE regionid LIKE '%_8' OR regionid LIKE '%_9';
    
    For query topology and execution plan please run: EXPLAIN <QueryId>
    
    Local runtime statistics
    ------------------------
    messages-per-sec:      2.01   total-messages:     10515     last-message: 3/14/18 2:25:40 PM PDT
     failed-messages:         0 failed-messages-per-sec:         0      last-failed:       n/a
    (Statistics of the local KSQL server interaction with the Kafka topic pageviews_enriched_r8_r9)
    
  • Discover the query execution plan with the EXPLAIN command. For example, run this command to view the query execution plan for CTAS_PAGEVIEWS_REGIONS:

    EXPLAIN CTAS_PAGEVIEWS_REGIONS;
    

    Your should look like this:

    Type                 : QUERY
    SQL                  : CREATE TABLE pageviews_regions AS SELECT gender, regionid , COUNT(*) AS numusers FROM pageviews_female WINDOW TUMBLING (size 30 second) GROUP BY gender, regionid HAVING COUNT(*) > 1;
    
    
    Local runtime statistics
    ------------------------
    messages-per-sec:      1.42   total-messages:     13871     last-message: 3/14/18 2:50:02 PM PDT
     failed-messages:         0 failed-messages-per-sec:         0      last-failed:       n/a
    (Statistics of the local KSQL server interaction with the Kafka topic PAGEVIEWS_REGIONS)
    
    Execution plan
    --------------
     > [ PROJECT ] Schema: [GENDER : STRING , REGIONID : STRING , NUMUSERS : INT64].
             > [ FILTER ] Schema: [PAGEVIEWS_FEMALE.GENDER : STRING , PAGEVIEWS_FEMALE.REGIONID : STRING , PAGEVIEWS_FEMALE.ROWTIME : INT64 , KSQL_AGG_VARIABLE_0 : INT64 , KSQL_AGG_VARIABLE_1 : INT64].
                     > [ AGGREGATE ] Schema: [PAGEVIEWS_FEMALE.GENDER : STRING , PAGEVIEWS_FEMALE.REGIONID : STRING , PAGEVIEWS_FEMALE.ROWTIME : INT64 , KSQL_AGG_VARIABLE_0 : INT64 , KSQL_AGG_VARIABLE_1 : INT64].
                             > [ PROJECT ] Schema: [PAGEVIEWS_FEMALE.GENDER : STRING , PAGEVIEWS_FEMALE.REGIONID : STRING , PAGEVIEWS_FEMALE.ROWTIME : INT64].
                                     > [ SOURCE ] Schema: [PAGEVIEWS_FEMALE.ROWTIME : INT64 , PAGEVIEWS_FEMALE.ROWKEY : STRING , PAGEVIEWS_FEMALE.USERID : STRING , PAGEVIEWS_FEMALE.PAGEID : STRING , PAGEVIEWS_FEMALE.REGIONID : STRING , PAGEVIEWS_FEMALE.GENDER : STRING].
    
    
    Processing topology
    -------------------
    Topologies:
       Sub-topology: 0
        Source: KSTREAM-SOURCE-0000000000 (topics: [PAGEVIEWS_FEMALE])
          --> KSTREAM-MAPVALUES-0000000001
        Processor: KSTREAM-MAPVALUES-0000000001 (stores: [])
          --> KSTREAM-TRANSFORMVALUES-0000000002
          <-- KSTREAM-SOURCE-0000000000
        ...
    
      Sub-topology: 1
        Source: KSTREAM-SOURCE-0000000008 (topics: [KSQL_Agg_Query_1521052072079-repartition])
          --> KSTREAM-AGGREGATE-0000000005
        Processor: KSTREAM-AGGREGATE-0000000005 (stores: [KSQL_Agg_Query_1521052072079])
          --> KTABLE-FILTER-0000000009
          <-- KSTREAM-SOURCE-0000000008
        ...
    

For more information about KSQL syntax, see Syntax Reference.

Next Steps

Learn more about the components shown in this quick start: