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, Avro, or Protobuf. In this guide, we show you how to read each of these formats into Deephaven tables, and how to write a Deephaven table to Kafka.

See A Kafka introduction: basic terms for a detailed discussion of Kafka topics and supported formats. See the Apache Kafka documentation for full details on how to use Kafka.

Key and value

Each Kafka record has a key and a value. Kafka writes each record to a partition at an offset, with a timestamp. For example, a list of Kafka messages might have a stock ticker as the key and its price as the value.

The key and value are similar in that they can be nearly any sequence of bytes. The primary difference is that Kafka's default partitioner hashes a non-null key to choose a partition, so all records with the same key go to the same partition as long as the topic's partition count doesn't change.

A key spec and a value spec tell Deephaven how to turn the key and the value into table columns.

When a single-column key spec doesn't name its column, the name comes from the deephaven.key.column.name consumer property, an entry in the configuration dictionary passed as the first argument to consume. If that property isn't set, the name defaults to KafkaKey. Value specs work the same way with deephaven.value.column.name and KafkaValue.

Deephaven chooses the column type in this order:

1. The type set in the spec, if there is one.
2. The deephaven.key.column.type or deephaven.value.column.type property. It accepts short, int, long, float, double, byte[], or String (also accepted as string).
3. The Kafka deserializer set in the key.deserializer or value.deserializer consumer property. Deephaven recognizes the numeric, byte-array, UUID, ByteBuffer, and Bytes deserializers.

For a String column, set the type property or pass the type to the spec.

The key and the value can each be read as:

• simple type
• JSON encoded
• Avro encoded
• Protobuf encoded
• parsed by an object processor, such as a Jackson JSON provider
• ignored

You can't ignore both the key and the value.

Standard data fields

Besides the key and value, each record's partition, offset, and timestamp become columns in the table that Deephaven creates when it consumes the topic. You can change the column names, but not the column types. You can also add optional columns for the receive time and for the key and value sizes. The receive time is the date-time immediately after Deephaven observes the record.

Consumer properties control these columns. These are entries in the configuration dictionary passed as the first argument to consume. To add an optional column, set its property to the column name you want. To disable a column that is present by default, set its property to an empty string. The property doesn't accept a null value. For example, you might disable the partition column when the topic has only one partition.

Table types

Deephaven Kafka tables can be append-only, blink, or ring. Set the type with the table_type argument, using the TableType class from deephaven.stream.kafka.consumer, imported as kc in the examples on this page.

• Append-only tables keep every row. The table and its memory use can grow without limit. To use this type, pass table_type=kc.TableType.append().
• Blink tables keep only the rows from the current update cycle. Each new message appears as a row for one update cycle and then disappears. Blink is the default table type. To set it explicitly, use table_type=kc.TableType.blink().
• Ring tables keep only the last N rows. When the table grows beyond N rows, it discards the oldest rows until N remain. To use this type, pass table_type=kc.TableType.ring(N).

Combine a blink table with a stateful aggregation such as last_by to keep results after the rows disappear.

Launching Kafka with Deephaven

Deephaven has an official Docker Compose file that contains the Deephaven images along with a Redpanda image. Redpanda lets you input data directly into a Kafka stream from the terminal. Redpanda is one of many Kafka-compatible event streaming platforms that work with Deephaven.

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

Consume a Kafka stream

In this example, we consume a Kafka topic (test.topic) as a Deephaven table. You populate the Kafka topic by entering commands in the terminal.

For demonstration purposes, we use an append-only table and ignore the Kafka key.

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

The table_type argument, kc.TableType.append(), creates an append-only table. The key_spec argument, kc.KeyValueSpec.IGNORE, ignores the Kafka key. The value_spec argument, simple_spec ("Command", dht.string), reads each record's value into a String column named Command.

The result_append table is now subscribed to all partitions in the test.topic topic. When you send data to the test.topic topic, it appears in the table.

Input Kafka data for testing

For this example, you enter information into the Kafka topic from a terminal. To do this, run:

This waits for input from the terminal and sends each line you enter to the test.topic topic as a record. Press Ctrl + D when you're done to stop producing.

Once sent, that information appears automatically in your Deephaven table.

Ring and blink tables

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

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 the data, the result_ring table contains the last three entries, and the result_blink table shows each new row for only one update cycle.

Because rows disappear from the result_blink table after one update cycle, let's add a table that uses last_by to keep the last row added to the result_blink table.

Read the key and choose partitions and offsets

In this example, consume reads the Kafka topic share.price into an append-only table. Unlike the previous examples, it reads both the key and the value, and it sets the partitions and the starting offsets explicitly.

When reading a Kafka topic, you can select which partitions to listen to. By default, Deephaven reads all partitions. You can also choose where reading starts in each partition: at the beginning, at the end, at a specific offset, or without seeking. By default, Deephaven doesn't seek. Reading then starts where Kafka's consumer settings say. The offsets bullet below explains those settings.

Let's walk through the arguments in this query.

• 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.
• offsets is set to kc.ALL_PARTITIONS_DONT_SEEK, which doesn't seek. If the consumer's consumer group, set with the group.id property, has a committed offset, the consumer starts there. Otherwise, Kafka's auto.offset.reset property decides where to start. Its default, latest, reads only new messages.
• key_spec is set to simple_spec ('Symbol', dht.string), 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_spec('Price', dht.string), which instructs the consumer to expect messages with a Kafka value field, and creates a Price column of type String to store the information. Deephaven reads the price as a string because Redpanda's command-line tool, rpk, sends it as text. A numeric type such as dht.double would use Kafka's binary deserializer, which expects binary values.
• table_type is set to kc.TableType.append(), which creates an append-only table.

To choose a different starting point, pass one of these values as offsets:

• kc.ALL_PARTITIONS_SEEK_TO_END always starts with new messages only.
• kc.ALL_PARTITIONS_SEEK_TO_BEGINNING starts at the beginning of every partition.
• A dictionary that maps partition numbers to offsets, such as {0: 100, 1: 250}, starts at those offsets. Partitions that the dictionary doesn't list don't seek.

To listen to specific partitions, pass them as a list of integers, such as partitions=[1, 3, 5].

Now let's add some entries to our Kafka stream.

Run the following command:

Enter as many key-value pairs as you want. Put a space between each key and its value, and put each pair on its own line:

To drop the Symbol column, pass key_spec=kc.KeyValueSpec.IGNORE instead, as in the first example.

Read Kafka topic in JSON format

The following two examples read a Kafka topic called orders in JSON format.

This example uses json_spec:

Here, the value_spec argument uses json_spec, which parses each record's value as JSON.

The first argument to json_spec is a table definition: a dictionary that maps each column name in the result table to its Deephaven type (for example, dht.double).

The mapping keyword argument of json_spec is a dictionary that maps JSON field names to table column names. Each column name must appear in the table definition from the first argument. 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" reads the JSON field price into the Price column of the result table. For a column that the map doesn't mention, Deephaven reads the JSON field with the same name as the column.

If you omit mapping, Deephaven assumes that JSON field names match column names.

This example uses object_processor_spec with a Jackson provider:

Here, object_processor_spec takes a Jackson provider built from a dictionary that maps each JSON field name to its type. The resulting columns take the JSON field names, so the example renames them with rename_columns.

To test either example, run the following command:

Then enter the following values:

Read Kafka topic in Avro format

In this example, consume reads the Kafka topic share.price in Avro format. This example assumes that a schema named share.price.record is registered in the schema registry of the Redpanda instance from the Docker Compose file.

A schema registry stores Kafka event schema definitions and tracks their versions so that producers and consumers can share them. To register a schema, see Redpanda's schema registry documentation.

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 a JSON-encoded Avro schema definition string.

avro_spec also takes three optional keyword arguments:

• schema_version is the version of the named schema to get from the schema registry. The default, latest, gets the latest available version.
• mapping is a dictionary that maps Avro field names to table column names. Unless mapped_only is True, Deephaven maps each Avro field that mapping doesn't name to a column of the same name.
• mapped_only is a boolean that defaults to False. When it is True and mapping is set, Deephaven leaves Avro fields that mapping doesn't name out of the resulting table. Without mapping, it has no effect.

Read Kafka topic in Protobuf format

In this example, consume reads the Kafka topic share.price in Protobuf format. Protobuf is Google's open-source, language-neutral format for serializing structured data.

This example assumes that a schema with the subject name share.price.record is registered in the schema registry of the Redpanda instance from the Docker Compose file. A schema registry stores each schema under a subject name and tracks the subject's versions. To register a schema, see Redpanda's schema registry documentation.

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 protobuf_spec, which specifies a Protobuf format for the Kafka value field.

protobuf_spec takes the following arguments:

• schema is the schema subject name, share.price.record in the example. When set, Deephaven fetches the Protobuf message descriptor from the schema registry. Set either this or message_class, but not both.
• message_class is the fully qualified Java class name for the Protobuf message on the current classpath, for example com.example.MyMessage or com.example.OuterClass$MyMessage. When this is set, Deephaven doesn't use the schema registry. Set either this or schema, but not both.
• schema_version specifies the schema version, or None (the default) for latest. In cases where restarts cause schema changes, we recommend setting this to ensure the resulting table definition doesn't change.
• schema_message_name is the fully qualified Protobuf message name, for example com.example.MyMessage. This message's descriptor is the basis for the resulting table's definition. If None (the default), Deephaven uses the first message descriptor in the Protobuf schema. We recommend setting this explicitly.
• include is a list of /-separated field paths to include. End a path with /* to also include every field path that starts with it. A field is included when any path in the list matches it. The default, None, includes all fields.
  • For example, include=["/foo/bar"] includes the field bar inside foo, along with its parents: the top-level message and foo.
  • include=["/foo/bar/*"] also includes every field nested under /foo/bar, such as /foo/bar/baz and /foo/bar/baz/zap.
• protocol is the wire protocol for this payload, as a kc.ProtobufProtocol.
  • When schema is set, the default is kc.ProtobufProtocol.serdes().
  • When message_class is set, the default is kc.ProtobufProtocol.raw().

Join and aggregate two Kafka streams

In this example, consume reads two Kafka topics, quotes and orders, into Deephaven as blink tables. The example uses last_by to track the latest data from each topic, natural_join to join the streams, and agg.sum_ and agg.weighted_sum to aggregate the results.

The quotes consumer doesn't pass key_spec. Instead, the deephaven.key.column.name and deephaven.key.column.type properties read each record's key into a String column named Symbol (see Key and value).

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

Then add the following entries:

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

Then add the following entries in the terminal:

The tables update as you add each entry to the Kafka streams. The final results are in the total_notional table.

Consume a Kafka stream into a partitioned table

Partitioned tables are Deephaven tables that are partitioned into subtables by one or more key columns. They have their own operations and are useful when working with large data sets.

To consume from Kafka directly into a partitioned table, use consume_to_partitioned_table. Its syntax is similar to that of consume. The result is always partitioned by Kafka partition. By default, it reads all partitions. To read specific partitions, pass them as a list with the partitions argument, such as partitions=[1, 3, 5], as in Read the key and choose partitions and offsets.

Custom Kafka parser

Some use cases call for custom parsing of Kafka streams, such as when payloads use non-standard encodings or need complex transformation.

See the dedicated guide, Write your own custom parser for Kafka, for a step-by-step walkthrough and complete examples.

Write to a Kafka stream

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

In this example, we write a simple time table to a topic called time-topic. The example writes column X as each record's key and ignores the value.

To see the records arrive, run docker compose exec redpanda rpk topic consume time-topic in a terminal.

Now we write a time table to a topic called time-topic_group. The last argument, True, sets last_by_key_columns, which tells the producer to perform a last_by on the key columns before writing to the stream.

Related documentation

• Kafka basic terminology
• Custom parser for Kafka
• consume
• consume_to_partitioned_table
• produce
• time_table
• last_by
• Kafka Pydoc