# Kafka

[Apache Kafka](http://kafka.apache.org/) is a distributed streaming platform. It allows you to create publishers, which create data streams, and consumers, which subscribe to and ingest the data streams produced by publishers.

You can use HeavyDB [KafkaImporter](https://docs.omnisci.com/latest/6_loading_data.html#kafkaimporter) C++ program to consume a topic created by running Kafka shell scripts from the command line. Follow the procedure below to use a Kafka producer to send data, and a Kafka consumer to store the data, in HeavyDB.

{% hint style="info" %}
This example assumes you have already installed and configured Apache Kafka. See the [Kafka website](https://kafka.apache.org/).
{% endhint %}

## Creating a Topic

Create a sample topic for your Kafka producer.

1. Run the `kafka-topics.sh` script with the following arguments:

   ```
   bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1
   --partitions 1 --topic matstream
   ```
2. Create a file named `myfile` that consists of comma-separated data. For example:

   ```
   michael,1
   andrew,2
   ralph,3
   sandhya,4
   ```
3. Use `heavysql` to create a table to store the stream.

   ```
   create table stream1(name text, id int);
   ```

## Using the Producer

Load your file into the Kafka producer.

1. Create and start a producer using the following command.

   ```
   cat myfile | bin/kafka-console-producer.sh --broker-list localhost:9097
   --topic matstream
   ```

## Using the Consumer

Load the data to HeavyDB using the Kafka console consumer and the `KafkaImporter` program.

1. Pull the data from Kafka into the `KafkaImporter` program.

   ```
   /home/heavyai/build/bin/KafkaImporter stream1 heavyai -p HyperInteractive -u heavyai --port 6274 --batch 1 --brokers localhost:6283  
   --topic matstream --group-id 1

   Field Delimiter: ,
   Line Delimiter: \n
   Null String: \N
   Insert Batch Size: 1
   1 Rows Inserted, 0 rows skipped.
   2 Rows Inserted, 0 rows skipped.
   3 Rows Inserted, 0 rows skipped.
   4 Rows Inserted, 0 rows skipped.
   ```
2. Verify that the data arrived using `heavysql`.

   ```
   heavysql> select * from stream1;
   name|id
   michael|1
   andrew|2
   ralph|3
   sandhya|4
   ```