Quick Start for Apache Kafka using Confluent Platform (Local)

Use this quick start to get up and running with Confluent Platform and its main components in a development environment. This quick start uses Confluent Control Center included in Confluent Platform for topic management and event stream processing using ksqlDB.

In this quick start, you create Apache Kafka® topics, use Kafka Connect to generate mock data to those topics, and create ksqlDB streaming queries on those topics. You then go to Control Center to monitor and analyze the streaming queries.

See also

You can also run an automated version of this quick start designed for Confluent Platform local installs.

Prerequisites:

Step 1: Download and Start Confluent Platform

  1. Go to the downloads page.

  2. Select Confluent Platform and click DOWNLOAD FREE.

    Tip

    You can download a previous version from Previous Versions.

  3. Provide the following:

    • Email: Your email address
    • Deployment Type: Manual Deployment
    • Type: zip
  4. Click DOWNLOAD FREE.

  5. Decompress the file. You should have the directories, such as bin and etc.

  6. Set the environment variable for the Confluent Platform directory.

    export CONFLUENT_HOME=<path-to-confluent>
    
  7. Add the Confluent Platform bin directory to your PATH.

    export PATH=$PATH:$CONFLUENT_HOME/bin
    
  8. Install the Kafka Connect Datagen source connector using the Confluent Hub client. This connector generates mock data for demonstration purposes and is not suitable for production. Confluent Hub is an online library of pre-packaged and ready-to-install extensions or add-ons for Confluent Platform and Kafka.

    confluent-hub install \
       --no-prompt confluentinc/kafka-connect-datagen:latest
    
  9. Start Confluent Platform using the Confluent CLI confluent local services start command. This command starts all of the Confluent Platform components, including Kafka, ZooKeeper, Schema Registry, HTTP REST Proxy for Kafka, Kafka Connect, ksqlDB, and Control Center.

    Important

    The confluent local commands are intended for a single-node development environment and are not suitable for a production environment. The data that are produced are transient and are intended to be temporary. For production-ready workflows, see Install and Upgrade Confluent Platform.

    confluent local start
    

    Your output should resemble:

    Starting Zookeeper
    Zookeeper is [UP]
    Starting Kafka
    Kafka is [UP]
    Starting Schema Registry
    Schema Registry is [UP]
    Starting Kafka REST
    Kafka REST is [UP]
    Starting Connect
    Connect is [UP]
    Starting KSQL Server
    KSQL Server is [UP]
    Starting Control Center
    Control Center is [UP]
    

Step 2: Create Kafka Topics

In this step, you create Kafka topics using Confluent Control Center. Confluent Control Center provides the functionality for building and monitoring production data pipelines and event streaming applications.

  1. Navigate to the Control Center web interface at http://localhost:9021.

    If you installed Confluent Platform on a different host, replace localhost with the host name in the address.

    It may take a minute or two for Control Center to come online.

    Note

    Control Center won’t connect to ksqlDB if Control Center isn’t open and running in a localhost browser session.

  2. Select the CO Cluster 1 cluster.

    ../_images/c3-landing-page.png
  3. Click Topics from the cluster submenu and click + Add a topic.

    ../_images/c3-create-topic.png
  4. In the Topic name field, specify pageviews and click Create with defaults.

    ../_images/c3-create-topic-name.png
  5. Click Topics from the cluster submenu and click + Add a topic.

  6. In the Topic name field, specify users and click Create with defaults.

Step 3: Install a Kafka Connector and Generate Sample Data

In this step, you use Kafka Connect to run a demo source connector called kafka-connect-datagen that creates sample data for the Kafka topics pageviews and users.

Tip

The Kafka Connect Datagen connector was installed manually in Step 1: Download and Start Confluent Platform. If you encounter issues locating the Datagen Connector, refer to the Issue: Cannot locate the Datagen connector in the Troubleshooting section.

  1. Run the first instance of the Kafka Connect Datagen connector to produce Kafka data to the pageviews topic in AVRO format.

    1. Select the CO Cluster 1 cluster.

    2. Click Connect.

    3. Select the connect-default cluster in the All Connect Clusters table.

    4. Click Add connector.

    5. Select the DatagenConnector tile.

      Tip

      To narrow displayed connectors, click Filter by type and click Sources.

    6. In the Name field, specify datagen-pageviews.

    7. After naming the connector, new fields appear. Scroll down and specify the following configuration values:

      • In Key converter class, specify org.apache.kafka.connect.storage.StringConverter.
      • In kafka.topic, specify pageviews.
      • In max.interval, specify 100.
      • In quickstart, specify pageviews.
    8. Click Continue.

    9. Review the connector configuration and click Launch.

      ../_images/connect-review-pageviews.png
  2. Run the second instance of the Kafka Connect Datagen connector to produce Kafka data to the users topic in AVRO format.

    1. Select the CO Cluster 1 cluster.

    2. Click Connect.

    3. Select the connect-default cluster in the All Connect Clusters table.

    4. Click + Add connector.

    5. Select the DatagenConnector tile.

      Tip

      To narrow displayed connectors, click Filter by type and click Sources.

    6. In the Name field, specify datagen-users.

    7. After naming the connector, new fields appear. Scroll down and specify the following configuration values:

      • In Key converter class, specify org.apache.kafka.connect.storage.StringConverter.
      • In kafka.topic, specify users.
      • In max.interval, specify 1000.
      • In quickstart, specify users.
    8. Click Continue.

    9. Review the connector configuration and click Launch.

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

Tip

You can also run these commands using the ksqlDB CLI from your terminal with this command: <path-to-confluent>/bin/ksql http://localhost:8088.

Create Streams and Tables

In this step, you use ksqlDB to create a stream for the pageviews topic and a table for the users topic.

  1. Select the CO Cluster 1 cluster.
  2. Click ksqlDB.
  3. Select the ksqlDB application.
  4. From the ksqlDB EDITOR page, in the Streams tab, click + Add Stream.
  5. Select the pageviews topic.
  6. Choose your stream options:
    • In Value format, select AVRO.
    • In Value column(s), set the fields as follows:
      • viewtime with type BIGINT
      • userid with type VARCHAR
      • pageid with type VARCHAR
  7. Click Save STREAM.
  8. In the Tables tab, click Add a table.
  9. Select the users topic.
  10. Choose your table options:
    • In Value format, select AVRO.
    • In PRIMARY KEY column, select userid.
    • In the Value column(s) section, set the fields as follows:
      • registertime with type BIGINT
      • userid with type VARCHAR
      • regionid with type VARCHAR
      • gender with type VARCHAR
  11. Click Save TABLE.

Write Queries

In this step, you create ksqlDB queries against the stream and the tabe you created above.

  1. Select the CO Cluster 1 cluster.

  2. Click ksqlDB.

  3. Select the ksqlDB application.

  4. In the Editor tab, click Add query properties to add a custom query property.

  5. Set the auto.offset.reset parameter to Earliest.

    This setting instructs ksqlDB queries to read all available topic data from the beginning. This configuration is used for each subsequent query. For more information, see the ksqlDB Configuration Parameter Reference.

  6. Create the following queries.

    1. Create a non-persistent query that returns data from a stream with the results limited to a maximum of three rows:

      Enter the following query in the editor:

      SELECT pageid FROM pageviews EMIT CHANGES LIMIT 3;
      
    2. Click Run query. Your output should resemble:

      ../_images/c3-ksql-query-results-pageid.png

      Click the Card view or Table view icon to change the output layout.

    3. Create a persistent query (as a stream) that filters pageviews stream for female users. The results from this query are written to the Kafka PAGEVIEWS_FEMALE topic:

      Enter the following query in the editor:

      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';
      
    4. Click Run query. Your output should resemble:

      ../_images/c3-ksql-persist-query-pv-female-results.png
    5. Create a persistent query where the regionid ends with 8 or 9. Results from this query are written to a Kafka topic named pageviews_enriched_r8_r9:

      Enter the following query in the editor:

      CREATE STREAM pageviews_female_like_89
         WITH (kafka_topic='pageviews_enriched_r8_r9', value_format='AVRO')
         AS SELECT * FROM pageviews_female
         WHERE regionid LIKE '%_8' OR regionid LIKE '%_9';
      
    6. Click Run query. Your output should resemble:

      ../_images/c3-ksql-persist-query-pv-female89-results.png
    7. 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:

      Enter the following query in the editor:

      CREATE TABLE pageviews_regions
         AS SELECT gender, regionid , COUNT(*) AS numusers
         FROM pageviews LEFT JOIN users ON pageviews.userid = users.userid
         WINDOW TUMBLING (size 30 second)
         GROUP BY gender, regionid
         HAVING COUNT(*) > 1;
      
    8. Click Run query. Your output should resemble:

      ../_images/c3-ksql-persist-query-table-results.png
    9. Click the Running queries tab. You should see the following persisted queries:

      • PAGEVIEWS_FEMALE
      • PAGEVIEWS_FEMALE_LIKE_89
      • PAGEVIEWS_REGIONS
    10. Click the Editor tab. The All available streams and tables pane shows all of the streams and tables that you can access.

      ../_images/c3-ksql-stream-table-view-1.png
    11. In the All available streams and tables section, click KSQL_PROCESSING_LOG to view the stream’s schema, including nested data structures.

Run Queries

In this step, you run the ksqlDB queries you save as streams and tables above in the previous section.

  1. Select the CO Cluster 1 cluster.

  2. Click ksqlDB.

  3. Select the ksqlDB application.

  4. In the Streams tab, select the PAGEVIEWS_FEMALE stream.

  5. Click Query stream.

    Streaming output of the query displays.

  6. Click Stop to stop the output generation.

  7. In the Tables tab, select PAGEVIEWS_REGIONS table.

  8. Click Query table.

    Streaming output of the query displays.

  9. Click Stop to stop the output generation.

Step 5: Monitor Consumer Lag

  1. Select the CO Cluster 1 cluster.
  2. Click Consumers to view the consumers created by ksqlDB.
  3. Click the consumer group ID to view details for the _confluent-ksql-default_query_CSAS_PAGEVIEWS_FEMALE consumer group.
../_images/ksql-interface-monitor.png

From this page you can see the consumer lag and consumption values for your streaming query.

../_images/ksql-interface-monitor-cnsmgp.png

For more information, see the Control Center Consumers documentation.

Step 6: Stop Confluent Platform

When you are done working with the local install, you can stop Confluent Platform.

  1. Stop Confluent Platform using the Confluent CLI confluent local services connect stop command.

    confluent local services stop
    
  2. Destroy the data in the Confluent Platform instance with the confluent local destroy command.

    confluent local destroy
    

Troubleshooting

If you encountered any issues while going through the quickstart workflow, review the following resolutions before trying the steps again.

Issue: Cannot locate the Datagen connector

Resolution: Verify that you have added the location of the Confluent Platform bin directory to your PATH as described in Step 1: Download and Start Confluent Platform:

export PATH=<path-to-confluent>/bin:$PATH

Resolution: Verify the DataGen connector is installed and running.

Ensure that the kafka-connect-datagen is installed and running as described in Step 1: Download and Start Confluent Platform.

confluent-hub install --no-prompt confluentinc/kafka-connect-datagen:latest

Your output should resemble:

Running in a "--no-prompt" mode
...
Completed

Resolution: Check the connect logs for Datagen using the Confluent CLI confluent local services connect log command.

confluent local services connect log | grep -i Datagen

Your output should resemble:

[2019-04-18 14:21:08,840] INFO Loading plugin from: /Users/user.name/Confluent/confluent-version/share/confluent-hub-components/confluentinc-kafka-connect-datagen (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:215)
[2019-04-18 14:21:08,894] INFO Registered loader: PluginClassLoader{pluginLocation=file:/Users/user.name/Confluent/confluent-version/share/confluent-hub-components/confluentinc-kafka-connect-datagen/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:238)
[2019-04-18 14:21:08,894] INFO Added plugin 'io.confluent.kafka.connect.datagen.DatagenConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:167)
[2019-04-18 14:21:09,882] INFO Added aliases 'DatagenConnector' and 'Datagen' to plugin 'io.confluent.kafka.connect.datagen.DatagenConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:386)

Resolution: Verify the .jar file for kafka-connect-datagen has been added and is present in the lib subfolder.

ls <path-to-confluent>/share/confluent-hub-components/confluentinc-kafka-connect-datagen/lib/

Your output should resemble:

...
kafka-connect-datagen-0.1.0.jar
...

Resolution: Verify the plugin exists in the connector path.

When you installed the kafka-connect-datagen file from Confluent hub, the installation directory is added to the plugin path of several properties files:

 Adding installation directory to plugin path in the following files:
/Users/user.name/Confluent/confluent-version/etc/kafka/connect-distributed.properties
/Users/user.name/Confluent/confluent-version/etc/kafka/connect-standalone.properties
/Users/user.name/Confluent/confluent-version/etc/schema-registry/connect-avro-distributed.properties
/Users/user.name/Confluent/confluent-version/etc/schema-registry/connect-avro-standalone.properties
...

You can use any of them to check the connector path. This example uses the connect-avro-distributed.properties file.

grep plugin.path <path-to-confluent>/etc/schema-registry/connect-avro-distributed.properties

Your output should resemble:

plugin.path=share/java,/Users/user.name/Confluent/confluent-version/share/confluent-hub-components

Confirm its contents are present:

ls <path-to-confluent>/share/confluent-hub-components/confluentinc-kafka-connect-datagen

Your output should resemble:

assets   doc  lib  manifest.json

Issue: Stream-Stream joins error

An error states Stream-Stream joins must have a WITHIN clause specified. This error can occur if you created both pageviews and users as streams by mistake.

../_images/c3-ksql-stream-stream-join-error.png

Resolution: Ensure that you created a stream for pageviews, and a table for users in Step 4: Create and Write to a Stream and Table using ksqlDB.

Issue: Unable to successfully complete ksqlDB query steps

Java errors or other severe errors were encountered.

Resolution: Ensure you are on an Operating System currently supported by Confluent Platform.

Next Steps

Learn more about the components shown in this quick start: