Load data from Kafka

This tutorial refers to release artifacts that are may not be available in Imply 4.0. For best results, refer to the Druid ingestion documentation.

The Druid Kafka indexing service provides support for ingesting data from Kafka. This service offers exactly-once ingestion guarantees as well as the ability to ingest historical data. Additionally, this runs as part of the core Druid services and does not require any additional processes.

The Kafka indexing service, uses supervisors which run on the overlord and manage the creation and lifetime of Kafka indexing tasks. This indexing service can handle non-recent events and provides exactly-once ingestion semantics.

You can load data from Kafka in the Druid Console using the Apache Kafka data loader:

Tutorial: Loading data from Kafka

In this tutorial, you will load your own streams from Kafka using the Druid Kafka indexing service.

The following steps assume the use of Imply on-prem, but the general steps are similar for cloud Imply as well.

The steps also assume you have access to a running instance of Imply. If you don't, see the Quickstart for information on getting started.

Start Kafka

Apache Kafka is a high throughput message bus that works well with Druid. For this tutorial, we will use Kafka 2.1.0.

To download Kafka, run the following commands in your terminal:

curl -O https://archive.apache.org/dist/kafka/2.1.0/kafka_2.12-2.1.0.tgz
tar -xzf kafka_2.12-2.1.0.tgz
cd kafka_2.12-2.1.0

Start a Kafka broker by running the following command in a new terminal window:

./bin/kafka-server-start.sh config/server.properties

Run this command to create a Kafka topic called wikipedia, to which we will send data:

./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic wikipedia

Enable Druid Kafka ingestion

We will use Druid's Kafka indexing service to ingest messages from our newly created wikipedia topic. To start the service, we will need to submit a supervisor spec to the Druid overlord by running the following from the Imply directory:

curl -XPOST -H'Content-Type: application/json' -d @quickstart/wikipedia-kafka-supervisor.json http://localhost:8090/druid/indexer/v1/supervisor

If the supervisor was successfully created, you will get a response containing the ID of the supervisor; in our case we should see {"id":"wikipedia-kafka"}.

For more details about what's going on here, check out the Druid Kafka indexing service documentation.

Load historical data

Let's launch a console producer for our topic and send some data!

In your Kafka directory, run the following command, where {PATH_TO_IMPLY} is replaced by the path to the Imply directory:

export KAFKA_OPTS="-Dfile.encoding=UTF-8"
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic wikipedia < {PATH_TO_IMPLY}/quickstart/wikipedia-2016-06-27-sampled.json

The previous command posted sample events to the wikipedia Kafka topic, which were then ingested into Druid by the Kafka indexing service. You're now ready to run some queries!

Load real-time data

Exploring historical Wikipedia edits is interesting, but even more fascinating is being able to explore what is trending on Wikipedia right now. To do this, we will download a helper application which will parse events from Wikimedia's IRC feeds and post these events into the wikipedia Kafka topic we set up earlier.

curl -O https://static.imply.io/quickstart/wikiticker-0.8.tar.gz
tar -xzf wikiticker-0.8.tar.gz
cd wikiticker-0.8-SNAPSHOT

Now run wikiticker with parameters instructing it to write output to our Kafka topic:

bin/wikiticker -J-Dfile.encoding=UTF-8 -out kafka -topic wikipedia

Once the data is sent to Druid, you can immediately query it.

Query data

After sending data, you can immediately query it using any of the supported query methods including visualization and SQL.

To start off, try a SQL query:

$ bin/dsql
dsql> SELECT FLOOR(__time TO DAY) AS "Day", count(*) AS Edits FROM "wikipedia-kafka" GROUP BY FLOOR(__time TO DAY);

┌──────────────────────────┬───────┐
│ Day                      │ Edits │
├──────────────────────────┼───────┤
│ 2016-06-27T00:00:00.000Z │ 24433 │
│ 2017-03-07T00:00:00.000Z │   642 │
└──────────────────────────┴───────┘
Retrieved 2 rows in 0.04s.

You can see both the historical and real-time data loaded into the system. Next, try configuring a data cube in Imply:

Navigate to Pivot at http://localhost:9095.
Click on the Plus icon in the top right of the header bar and select "New data cube".
Select the source "druid: wikipedia-kafka" and ensure "Auto-fill dimensions and measures" is checked.
Click "Next: configure data cube".
Click "Create cube". You should see the confirmation message "Data cube created".
View your new data cube by clicking the Home icon in the top-right and selecting the "Wikipedia Kafka" cube you just created.

Load your own Kafka topics

So far, you've loaded data using a supervisor spec which we've included in the distribution. Each supervisor spec is designed to work with a particular dataset. You can load your own datasets by writing a custom ingestion spec.

To customize supervised Kafka indexing service ingestion, you can copy the included quickstart/wikipedia-kafka-supervisor.json spec into your own file, edit it as needed, and create or shutdown supervisors as needed. There's no need to restart Imply or Druid services themselves. See the Druid documentation for details on possible configurations.

When loading your own Kafka topics, Druid will create at least one segment for every Kafka partition, for every segmentGranularity period. If your segmentGranularity period is "HOUR" and you have 3 Kafka partitions, then Druid will
create at least one segment per hour. For best performance, we recommend avoiding creating too many segments by keeping
your number of Kafka partitions appropriately sized. See the
[Druid Kafka indexing service documentation](./druid/development/extensions-core/kafka-ingestion.md)
for more details.

Submitting a Supervisor spec

A supervisor spec can also be submitted via HTTP POST to http://<OVERLORD_IP>:<OVERLORD_PORT>/druid/indexer/v1/supervisor, for example:

curl -X POST -H 'Content-Type: application/json' -d @supervisor-spec.json http://localhost:8090/druid/indexer/v1/supervisor

A sample supervisor spec is shown below:

{
  "type": "kafka",
  "dataSchema": {
    "dataSource": "wikiticker-kafka",
    "parser": {
      "type": "string",
      "parseSpec": {
        "format": "json",
        "timestampSpec": {
          "column": "timestamp",
          "format": "auto"
        },
        "dimensionsSpec": {
          "dimensions": [
            "isRobot",
            "channel",
            "flags",
            "isUnpatrolled",
            "page",
            "diffUrl",
            "comment",
            "isNew",
            "isMinor",
            "user",
            "namespace",
            { "name" : "commentLength", "type" : "long" },
            { "name" : "deltaBucket", "type" : "long" },
            "cityName",
            "countryIsoCode",
            "countryName",
            "isAnonymous",
            "metroCode",
            "regionIsoCode",
            "regionName"
          ]
        }
      }
    },
    "metricsSpec" : [
      { "type" : "count", "name" : "count", "type" : "count" },
      { "type" : "longSum", "name" : "added", "fieldName" : "added" },
      { "type" : "longSum", "name" : "deleted", "fieldName" : "deleted" },
      { "type" : "longSum", "name" : "delta", "fieldName" : "delta" },
      { "type" : "hyperUnique", "name" : "user_unique", "fieldName" : "user" }
    ],
    "granularitySpec": {
      "type": "uniform",
      "segmentGranularity": "HOUR",
      "queryGranularity": "NONE",
      "rollup": true
    }
  },
  "ioConfig": {
    "topic": "wikiticker",
    "consumerProperties": {
      "bootstrap.servers": "myKafkaBroker:9092"
    }
  }
}

More information

Please refer to Druid's Kafka indexing service documentation for more details.