Tutorial: Load streaming data from Apache Kafka

    For this tutorial, we’ll assume you’ve already downloaded Druid as described in the using the single-machine configuration and have it running on your local machine. You don’t need to have loaded any data yet.

    Download and start Kafka

    is a high throughput message bus that works well with Druid. For this tutorial, we will use Kafka 2.7.0. To download Kafka, issue the following commands in your terminal:

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

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

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

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

    In your Druid directory, run the following command:

    1. cd quickstart/tutorial
    2. gunzip -c wikiticker-2015-09-12-sampled.json.gz > wikiticker-2015-09-12-sampled.json

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

    The previous command posted sample events to the wikipedia Kafka topic. Now we will use Druid’s Kafka indexing service to ingest messages from our newly created topic.

    Loading data with the data loader

    Navigate to localhost:8888 and click Load data in the console header.

    Select Apache Kafka and click Connect data.

    Data loader sample

    Enter localhost:9092 as the bootstrap server and wikipedia as the topic.

    Click Apply and make sure that the data you are seeing is correct.

    Once the data is located, you can click “Next: Parse data” to go to the next step.

    The data loader will try to automatically determine the correct parser for the data. In this case it will successfully determine json. Feel free to play around with different parser options to get a preview of how Druid will parse your data.

    With the json parser selected, click Next: Parse time to get to the step centered around determining your primary timestamp column.

    Data loader parse time

    Click Next: ... twice to go past the Transform and Filter steps. You do not need to enter anything in these steps as applying ingestion time transforms and filters are out of scope for this tutorial.

    In the Configure schema step, you can configure which and metrics will be ingested into Druid. This is exactly what the data will appear like in Druid once it is ingested. Since our dataset is very small, go ahead and turn off by clicking on the switch and confirming the change.

    Once you are satisfied with the schema, click Next to go to the Partition step where you can fine tune how the data will be partitioned into segments.

    Data loader partition

    Here, you can adjust how the data will be split up into segments in Druid. Since this is a small dataset, there are no adjustments that need to be made in this step.

    Click Next: Tune to go to the tuning step.

    In the Tune step is it very important to set Use earliest offset to True since we want to consume the data from the start of the stream. There are no other changes that need to be made here, so click Next: Publish to go to the Publish step.

    Data loader publish

    Let’s name this datasource wikipedia-kafka.

    Finally, click Next to review your spec.

    This is the spec you have constructed. Feel free to go back and make changes in previous steps to see how changes will update the spec. Similarly, you can also edit the spec directly and see it reflected in the previous steps.

    Once you are satisfied with the spec, click Submit and an ingestion task will be created.

    Tasks view

    You will be taken to the task view with the focus on the newly created supervisor.

    The task view is set to auto refresh, wait until your supervisor launches a task.

    When a tasks starts running, it will also start serving the data that it is ingesting.

    When the wikipedia-kafka datasource appears here it can be queried.

    Note: if the datasource does not appear after a minute you might have not set the supervisor to read from the start of the stream (in the step).

    At this point, you can go to the Query view to run SQL queries against the datasource.

    Since this is a small dataset, you can simply run a SELECT * FROM "wikipedia-kafka" query to see your results.

    Query view

    Check out the query tutorial to run some example queries on the newly loaded data.

    In the console, click Submit supervisor to open the submit supervisor dialog.

    Paste in this spec and click Submit.

    1. {
    2.   "type": "kafka",
    3.   "spec" : {
    4.     "dataSchema": {
    5.       "dataSource": "wikipedia",
    6.       "timestampSpec": {
    7.         "column": "time",
    8.         "format": "auto"
    9.       },
    10.       "dimensionsSpec": {
    11.         "dimensions": [
    12.           "cityName",
    13.           "comment",
    14.           "countryIsoCode",
    15.           "countryName",
    16.           "isAnonymous",
    17.           "isMinor",
    18.           "isNew",
    19.           "isRobot",
    20.           "isUnpatrolled",
    21.           "metroCode",
    22.           "namespace",
    23.           "page",
    24.           "regionIsoCode",
    25.           "regionName",
    26.           "user",
    27.           { "name": "added", "type": "long" },
    29.           { "name": "delta", "type": "long" }
    30.         ]
    31.       },
    32.       "metricsSpec" : [],
    33.       "granularitySpec": {
    34.         "type": "uniform",
    35.         "queryGranularity": "NONE",
    36.         "rollup": false
    37.       }
    38.     },
    39.     "tuningConfig": {
    40.       "type": "kafka",
    41.       "reportParseExceptions": false
    42.     },
    43.     "ioConfig": {
    44.       "topic": "wikipedia",
    45.       "inputFormat": {
    46.         "type": "json"
    47.       },
    48.       "replicas": 2,
    49.       "taskDuration": "PT10M",
    50.       "completionTimeout": "PT20M",
    51.       "consumerProperties": {
    52.         "bootstrap.servers": "localhost:9092"
    53.       }
    54.     }
    55.   }
    56. }

    This will start the supervisor that will in turn spawn some tasks that will start listening for incoming data.

    Submit a supervisor directly

    To start the service directly, we will need to submit a supervisor spec to the Druid overlord by running the following from the Druid package root:

    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"}.

    For more details about what’s going on here, check out the .

    You can view the current supervisors and tasks in the web console: http://localhost:8888/unified-console.md#tasks.

    After data is sent to the Kafka stream, it is immediately available for querying.

    Please follow the query tutorial to run some example queries on the newly loaded data.

    Cleanup

    To go through any of the other ingestion tutorials, you will need to shut down the cluster and reset the cluster state by removing the contents of the var directory in the Druid home, as the other tutorials will write to the same “wikipedia” datasource.

    You should additionally clear out any Kafka state. Do so by shutting down the Kafka broker with CTRL-C before stopping ZooKeeper and the Druid services, and then deleting the Kafka log directory at /tmp/kafka-logs: