Manage ksqlDB by using the Confluent CLI

In addition to the Cloud Console, you can use the Confluent CLI to create ksqlDB applications, create Kafka topics, and control access to cluster resources.

Install the Confluent Cloud CLI

After you have a working Kafka cluster in Confluent Cloud, you can use the Confluent CLI to interact with your cluster from your local computer. For example, you can produce and consume to a topic by using the Confluent CLI.

Install Confluent CLI as describe in CLI installation guide.

Tip

For examples of using multiple languages to connect Kafka client applications to Confluent Cloud, see the Code Examples.

Log in to your cluster

In the following steps, you log in to your Kafka cluster and connect to your cluster with an API key that you create by using the Confluent CLI.

For more information about Confluent CLI commands, see Confluent CLI Command Reference.

  1. Log in to your Confluent Cloud cluster.

    confluent login
    

    Your output should resemble:

    Enter your Confluent credentials:
    Email: jdoe@myemail.io
    Password: ***********************
    
    Logged in as "jdoe@myemail.io"
    Using environment "t118" ("default")
    
  2. View your cluster.

    confluent kafka cluster list
    

    Your output should resemble:

          Id      |       Name        |     Type     | Provider |  Region  | Availability | Status
    +-------------+-------------------+--------------+----------+----------+--------------+--------+
        lkc-emmox | ccloud-quickstart | BASIC_LEGACY | gcp      | us-west2 | HIGH         | UP
    
  3. Set the active Kafka cluster. In this example, the cluster ID is lkc-emmox.

    confluent kafka cluster use lkc-emmox
    

    Tip

    The lkc prefix in the cluster ID is an acronym for “logical Kafka cluster”.

Create a Kafka API key and secret

  1. Run the following to create an API key and secret. When you create an API key by using the CLI, it’s automatically stored locally.

    confluent api-key create --resource "cloud"
    
  2. Set the API key to use for Confluent CLI commands with the command confluent api-key use <key>.

    confluent api-key use --resource <resource-id> <api-key>
    

    For an API key/secret pair on the previously shown cluster with ID lkc-emmox, the command might resemble the following:

    confluent api-key use --resource lkc-emmox LD35EN2ZJTCTRQRM
    

Create a ksqlDB application by using the Confluent Cloud CLI

Use the ksql app create command to create a new ksqlDB application in your cloud-hosted Kafka cluster.

Run the following commmand to create a new ksqlDB application that has access to the same resources as the user account or service account that’s associated with the key.

confluent ksql app create <app-name> --api-key <api-key> --api-secret <secret>

If you don’t specify a Kafka API key, Confluent Cloud creates a service account for the application and a corresponding Kafka API key for the new account.

Create a topic by using the Confluent CLI

  1. Create a topic named users.

    confluent kafka topic create users
    

    The output from confluent kafka topic list should resemble:

         Name
    +------------+
        users
    

For more information, see confluent kafka topic command reference.

Enable ksqlDB access in Confluent Cloud

You can set ACLs to enable accessing the application by using the Confluent Cloud CLI.

  1. Configure the Confluent Cloud ACLs for ksqlDB. For more information about Confluent Cloud ACLs, see Use Access Control Lists (ACLs) for Confluent Cloud.

    1. In the Confluent Cloud CLI, find the ksqlDB application ID.

      confluent ksql app list
      

      Your output should resemble:

             Id      |   Name      | Topic Prefix |   Kafka   | Storage |                         Endpoint                          | Status
      +--------------+-------------+--------------+-----------+---------+-----------------------------------------------------------+--------+
        lksqlc-lg0g3 | ksqldb-app1 | pksqlc-4v5nn | lkc-emmox |     500 | https://pksqlc-emko7.us-central1.gcp.confluent.cloud:8088 | UP
      

      Tip

      The lksqlc and pksqlc prefixes are acronyms for “logical ksqlDB cluster” and “physical ksqlDB cluster”.

    2. Grant ksqlDB access to the users topic.

      confluent ksql app configure-acls <ksqldb-cluster-id> users --cluster <kafka-cluster-id>
      

      For the example clusters shown previously, the following command gives ksqlDB access to the users topic.

      confluent ksql app configure-acls lksqlc-lg0g3 users --cluster lkc-emmox
      

Create an API key for Confluent Cloud ksqlDB through the Confluent Cloud CLI

Starting with Confluent CLI v0.198.0, you can create API keys by using the Confluent CLI. For more information, see Create a resource-specific API key.

Important

The API key and secret that you create in this step are distinct from the key pair that you created when you installed the Confluent Cloud CLI. This key pair is for the ksqlDB application specifically and can only be created by using the confluent api-key create --resource <ksqldb-cluster-id> command.

Run the following command to see a list of all API keys associated with your ksqlDB application.

confluent api-key list --resource <ksqldb-cluster-id>

If the output is empty, follow the steps in Create a resource-specific API key.

Store an API key from another source

When you create an API key using the UI, the API, or with the CLI on another machine, the secret is not available for the CLI to use until you store it. This is required because you can’t retrieve secrets after creation. Add the API secret by using the confluent api-key store command.

From the Confluent Cloud CLI, type the following command:

confluent api-key store <api-key> <secret> --resource <cluster-id>

For an API key/secret pair on the example cluster with ID lkc-emmox, the command might resemble the following:

confluent api-key store LD35EN2ZJTCTRQRM 67JImN+9vk+Hj3eak2/UcwUlbDNlGGC3KAIOy5JNRVSnweumPBVpW31JWZSBeawz --resource lkc-emmox

Access a ksqlDB application in Confluent Cloud with an API key

ksqlDB in Confluent Cloud supports authentication with ksqlDB API keys.

Using ksqlDB CLI
To connect the ksqlDB CLI to a cluster, run the following command with your Confluent Cloud ksqlDB server URL specified.
$CONFLUENT_HOME/bin/ksql -u <ksql-api-key> -p <secret> <ccloud-ksql-server-url>
Using HTTPS Requests
Specify --basic in the Accept header of your request. Additionally, you need to send your ksqlDB API key and secret, separated by a colon, as the --user credentials.
  1. Run the confluent ksql app list command to get the URL of the ksqlDB endpoint.

    confluent ksql app list
    
           Id      |   Name      | Topic Prefix |   Kafka   | Storage |                         Endpoint                         | Status
    +--------------+-------------+--------------+-----------+---------+----------------------------------------------------------+--------+
      lksqlc-lg0g3 | ksqldb-app1 | pksqlc-4v5nn | lkc-emmox |     500 | https://pksqlc-emko7.us-central1.gcp.confluent.cloud:443 | UP
    
  2. Run the curl command to send a POST request to the ksql endpoint:

    curl -X "POST" "https://<cloud-ksqldb-endpoint>:443/ksql" \
         -H "Accept: application/vnd.ksql.v1+json" \
         --basic --user "<ksqldb-api-key>:<secret>" \
         -d $'{
      "ksql": "LIST STREAMS;",
      "streamsProperties": {}
    }'
    

    An example command for the previous ksqlDB application might resemble:

    curl -X "POST" "https://pksqlc-4v5nn.us-west2.gcp.confluent.cloud:443/ksql" \
         -H "Accept: application/vnd.ksql.v1+json" \
         --basic --user "7DOQ6QCUWCE6M0HD:25UCfE+p25wPXXHMJcIBg3dANJvr76LG4VZ5Hold+X2dVD5drvm8CdDaVdHia8d7" \
         -d $'{
      "ksql": "LIST STREAMS;",
      "streamsProperties": {}
    }'
    

    Your output should resemble:

    [
      {
        "@type": "streams",
        "statementText": "LIST STREAMS;",
        "streams": [
          {
            "type": "STREAM",
            "name": "KSQL_PROCESSING_LOG",
            "topic": "pksqlc-4v5nn-processing-log",
            "format": "JSON"
          },
          {
            "type": "STREAM",
            "name": "PAGEVIEWS_ENRICHED",
            "topic": "pksqlc-4v5nnPAGEVIEWS_ENRICHED",
            "format": "JSON"
          },
          {
            "type": "STREAM",
            "name": "PAGEVIEWS_ORIGINAL",
            "topic": "pageviews",
            "format": "JSON"
          }
        ],
        "warnings": []
      }
    ]
    
  3. Run the following command to inspect the streaming data in the PAGEVIEWS_ENRICHED stream. Press Ctrl+C to stop the query.

    curl -X "POST" "https://<cloud-ksqldb-endpoint>:443/query-stream" \
         --basic --user "<ksqldb-api-key>:<secret>" \
          -d $'{
       "sql": "SELECT * FROM PAGEVIEWS_ENRICHED EMIT CHANGES;",
       "streamsProperties": {}
     }'
    

    Your output should resemble:

    {"queryId":"d14df2c1-5995-4855-b300-f7710464c81c","columnNames":["USERID","PAGEID","REGIONID","GENDER"],"columnTypes":["STRING","STRING","STRING","STRING"]}
    ["User_7","Page_85","Region_8","FEMALE"]
    ["User_9","Page_47","Region_7","MALE"]
    ["User_4","Page_98","Region_7","OTHER"]
    ^C
    

See also

For an example that shows fully-managed Confluent Cloud connectors in action with Confluent Cloud ksqlDB, see the Cloud ETL Demo. This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud.

../_images/topology.png