Skip to main content
Version: Python

How to connect to a Kafka stream

Kafka is a distributed event streaming platform that lets you read, write, store, and process events, also called records.

Kafka topics take on many forms, such as raw input, JSON, or AVRO. In this guide, we show you how to import each of these formats as Deephaven tables.

Please see our concept guide, Kafka basic terminology, for a detailed discussion of Kafka topics and supported formats. See the Apache Kafka Documentation for full details on how to use Kafka.

Kafka in Deephaven

Standard data fields

Kafka has the standard data fields of partition, offset, and timestamp. Each of these fields becomes a column in the new table that stores the Kafka stream. The column names can be changed, but the type of column is set. The standard names and types for these values are:

  • KafkaPartition: int
  • KafkaOffset: long
  • KafkaTimestamp: DateTime

When reading a Kafka topic, you can select which partitions to listen to. By default, all partitions are read. Additionally, topics can be read from the beginning, from the latest offset, or from the first unprocessed offset. By default, all partitions are read from the latest offset.


The Kafka infrastructure can retain old messages for a maximum given age or retain the last message for individual keys.

While these three fields are traditionally included in the new table, you can choose to ignore them, such as when there is only one partition.

Key and value

The actual data of the Kafka stream are stored in the KafkaKey and KafkaValue columns. This is the information that is logged onto the partition with an offset at a certain time. For example, a list of Kafka messages might have a key of the user and a value of the message, and is logged at a certain time.

KafkaKey and KafkaValue are similar in that they can be nearly any sequence of bytes. The primary difference is that the key is used to create a hash that will facilitate load balancing. By default, each key and value are stored with column names of either KafkaKey or KafkaValue, and String type.

The KafkaKey and KafkaValue attributes can be:

  • simple type
  • JSON encoded
  • Avro encoded
  • ignored (cannot ignore both key and value)

Table types

Deephaven Kafka tables can be append-only, stream, or ring.

  • Append-only tables add one row for every message ingested — thus, table size and memory consumption can grow rapidly. Set this value with table_type = TableType.append().
  • Stream tables only keep the set of rows received during the last update cycle. This forms the basis for more advanced use cases when used in combination with stateful table aggregations like last_by. For streaming tables without any downstream table operations, aggregations or listeners, the new messages will appear as rows in the table for one UGP cycle, then disappear on the next UGP cycle. A streaming table is the default. You can set this value to table_type = to be explicit, but this is not required.
  • Ring tables retain the last N added rows for each update of the UGP cycle. If more than N rows are added with an update, the last N are stored in the ring table. Set this value with table_type = TableType.ring(N).

Launching Kafka with Deephaven

Deephaven has an official docker-compose file that contains the Deephaven images along with images from Redpanda. Redpanda allows us to input data directly into a Kafka stream from the terminal. This is just one of the supported Kafka-compatible event streaming platforms. Many more are available.

Save this locally as a docker-compose.yml file, and launch with docker-compose up.

Create a Deephaven table that listens to a Kafka topic

In this example, we consume a Kafka topic (test.topic) as a Deephaven table. The Kafka topic is populated by commands entered into the terminal.

For demonstration purposes, we will be using an append table and ignoring the Kafka key.

from import consumer as kc

result_append = kc.consume({'bootstrap.servers': 'redpanda:29092'},
'test.topic', table_type=kc.TableType.append(),

In this example, consume creates a Deephaven table from a Kafka topic. Here, {'bootstrap.servers': 'redpanda:29092'} is a dictionary describing how the Kafka infrastructure is configured. bootstrap.servers provides the initial hosts that a Kafka client uses to connect. In this case, bootstrap.servers is set to redpanda:29092.

table_type is set to kc.TableType.append() to create an append-only table, and key_spec is set to kc.KeyValueSpec.IGNORE to ignore the Kafka key.

The result table is now subscribed to all partitions in the test.topic topic. When data is sent to the test.topic topic, it will appear in the table.

Input Kafka data for testing

For this example, information is entered into the Kafka topic via a terminal. To do this, run:

docker-compose exec redpanda rpk topic produce test.topic

This will wait for input from the terminal and will send any input to the test.topic topic. Enter the information and use the keyboard shortcut Ctrl + D to send.

Once sent, that information will automatically appear in your Deephaven table.

Ring and stream tables

The following example shows how to create ring and stream tables to read from the test.topic topic:

result_ring = kc.consume({'bootstrap.servers': 'redpanda:29092'},
'test.topic', table_type=kc.TableType.ring(3),

result_stream = kc.consume({'bootstrap.servers': 'redpanda:29092'},

Let's run a few more docker-compose exec redpanda rpk topic produce test.topic commands to input additional data into the Kafka stream. As you can see, the result_append table contains all of the data, the result_ring table contains the last 3 entries, and the result_stream only shows rows before the next table update cycle is executed.

Since the result_stream table doesn't show the values in the topic, let's add a table to store the last value added to the result_stream table.

last_stream = result_stream.last_by()

Import a Kafka stream with append

In this example, consume reads the Kafka topic share.price. The specific key and value result in a table that appends new rows.

from import consumer as kc
import deephaven.dtypes as dht

result = kc.consume({'bootstrap.servers' : 'redpanda:29092'},

Let's walk through this query, focusing on the new optional arguments we've set.

  • partitions is set to None, which specifies that we want to listen to all partitions. This is the default behavior if partitions is not explicitly defined. To listen to specific partitions, we can define them as a list of integers (e.g., partitions=[1, 3, 5]).
  • offsets is set to ALL_PARTITIONS_DONT_SEEK, which only listens to new messages produced after this call is processed.
  • key_spec is set to simple('Symbol'), which instructs the consumer to expect messages with a Kafka key field, and creates a Symbol column of type String to store the information.
  • value_spec is set to simple('Price'), which instructs the consumer to expect messages with a Kafka value field, and creates a Price column of type String to store the information.
  • table_type is set to append, which creates an append-only table.

Now let's add some entries to our Kafka stream. Run docker-compose exec redpanda rpk topic produce share.price -f '%k %v\n' and enter as many key-value pairs as you want, separated by spaces and new lines:

AAPL 135.60
AAPL 135.99
AAPL 136.82

Import a Kafka stream ignoring keys

In this example, consume reads the Kafka topic share.price and ignores the partition and key values.

Run the same docker-compose exec redpanda rpk topic produce share.price -f '%k %v\n' command from the previous section and enter the sample key-value pairs.

from import consumer as kc
import deephaven.dtypes as dht

result = kc.consume({'bootstrap.servers' : 'redpanda:29092'},

As you can see, the key column is not included in the output table.

Read Kafka topic in JSON format

In this example, consume reads the Kafka topic orders in JSON format.

from import consumer as kc
import deephaven.dtypes as dht

result = kc.consume({ 'bootstrap.servers' : 'redpanda:29092' },
('Symbol', dht.string),
('Price', dht.double),
('Qty', dht.int_)],
'symbol' : 'Symbol',
'price' : 'Price',
'qty' : 'Qty'}

Run docker-compose exec redpanda rpk topic produce orders -f "%v\n" in your terminal and enter the following values:

{"symbol": "AAPL", "price": 135, "qty": 5}
{"symbol": "TSLA", "price": 730, "qty": 3}

In this query, the value_spec argument uses json_spec . A JSON parameterization is used for the KafkaValue field.

After this, we see an ordered list of Python tuples specifying column definitions.

  • The first element in each tuple is a string for the column name in the result table.
  • The second element in each tuple is a string for the column data type in the result table.

Within the value_spec argument, the keyword argument of mapping is given. This is a Python dictionary specifying a mapping from JSON field names to resulting table column names. Column names should be in the list provided in the first argument described above. The mapping dictionary may contain fewer entries than the total number of columns defined in the first argument.

In the example, the map entry 'price' : 'Price' specifies the incoming messages are expected to contain a JSON field named price, whose value will be mapped to the Price column in the resulting table. The columns not mentioned are mapped from matching JSON fields.

If the mapping keyword argument is not provided, it is assumed that JSON field names and column names will match.

Read Kafka topic in Avro format

In this example, consume reads the Kafka topic share.price in Avro format. This example uses an external schema definition registered in the development testing Redpanda instance that can be seen below. A Kafka Schema Registry allows sharing and versioning of Kafka event schema definitions.

from import consumer as kc

result = kc.consume({ 'bootstrap.servers' : 'redpanda:29092',
'schema.registry.url' : 'http://redpanda:8081' },
value_spec=kc.avro_spec('share.price.record', schema_version='1'),

In this query, the first argument includes an additional entry for schema.registry.url to specify the URL for a schema registry with a REST API compatible with Confluent's schema registry specification .

The value_spec argument uses avro_spec , which specifies an Avro format for the Kafka value field.

The first positional argument in the avro_spec call specifies the Avro schema to use. In this case, avro_spec gets the schema named share.price.record from the schema registry. Alternatively, the first argument can be an org.apache.avro.Schema object obtained from getAvroSchema .

Three optional keyword arguments are supported:

  • schema_version specifies the version of the schema to get, for the given name, from the schema registry. If not specified, the default of latest is assumed. This will retrieve the latest available schema version.
  • mapping expects a dictionary value, and if provided, specifies a name mapping for Avro field names to table column names. Any Avro field name not mentioned is mapped to a column of the same name.
  • mapping_only expects a dictionary value, and if provided, specifies a name mapping for Avro field names to table column names. Any Avro field name not mentioned is omitted from the resulting table.
  • When mapping and mapping_only are both omitted, all Avro schema fields are mapped to columns using the field name as column name.

Perform multiple operations

In this example, consume reads two Kafka topic, quotes and orders, into Deephaven as stream tables. Table operations are used to track the latest data from each topic (using last_by), join the streams together (natural_join), and aggregate (agg.sum_) the results.

from import consumer as kc
import deephaven.dtypes as dht

price_table = kc.consume(
{ 'bootstrap.servers' : 'redpanda:29092',
'' : 'Symbol'},
value_spec=kc.json_spec([('Price', dht.double)]))

last_price = price_table.last_by(by=["Symbol"])

orders_stream = kc.consume(
{ 'bootstrap.servers' : 'redpanda:29092' },
value_spec=kc.json_spec([('Symbol', dht.string),
('Id', dht.string),
('LimitPrice', dht.double),
('Qty', dht.int_) ]),,

orders_with_current_price = orders_stream \
.last_by('Id') \
.natural_join(table=last_price, on=["Symbol"], joins=["LastPrice = Price"])

from deephaven import agg

agg_list = [
agg.sum_("Shares = Qty"),
agg.weighted_sum("Qty", "Notional = LastPrice")

total_notional = orders_with_current_price.agg_by(agg_list, by = ["Symbol"])

Next, let's add records to the two topics. In a terminal, first run the following command to start writing to the quotes topic:

docker-compose exec redpanda rpk topic produce quotes -f '%k %v\n'

and add the following entries:

AAPL {"Price": 135}
AAPL {"Price": 133}
TSLA {"Price": 730}
TSLA {"Price": 735}

After submitting the entries to the quotes topic, use the following command t write to the orders topic:

docker-compose exec redpanda rpk topic produce orders -f "%v\n"

Then add the following entries in the terminal:

{"Symbol": "AAPL", "Id":"o1", "LimitPrice": 136, "Qty": 7}
{"Symbol": "AAPL", "Id":"o2", "LimitPrice": 132, "Qty": 2}
{"Symbol": "TSLA", "Id":"o3", "LimitPrice": 725, "Qty": 1}
{"Symbol": "TSLA", "Id":"o4", "LimitPrice": 730, "Qty": 9}

The tables will update as each entry is added to the Kafka streams. The final results are in the total_notional table.

Write to a Kafka stream

Deephaven can write tables to Kafka streams as well. When data in a table changes with real-time updates, those changes are also written to Kafka. The Kafka producer module defines methods to do this.

In this example, we write a simple time table to a topic called time-topic. With only one data point, we use the X as a key and ignore the value.

from deephaven import time_table
from deephaven import kafka_producer as pk
from import KeyValueSpec

source = time_table('00:00:00.1').update(formulas = ["X = i"])

write_topic = pk.produce(source, {'bootstrap.servers': 'redpanda:29092'},\
'time-topic', pk.simple_spec('X'), KeyValueSpec.IGNORE)

Now we write a time table to a topic called time-topic_group. The last argument is True for last_by_key_columns, which indicates we want to perform a last_by on the keys before writing to the stream.

import random
source_group = time_table('00:00:00.1').update(formulas = ["X = random.randint(1, 5)", "Y = i"])

write_topic_group = pk.produce(source_group, {'bootstrap.servers': 'redpanda:29092'},\
'time-topic_group',pk.json_spec(['X']), pk.json_spec(['X','Y',]), True)