Kafka write task

Send the samples of Synnax channels to a Kafka topic as JSON records.

A Kafka write task sends each sample of its channels to one topic as a JSON record. Use it to give other systems on Kafka the data that Synnax records. Before you start, connect a Kafka cluster.

The task reads channels and does not command hardware, so it takes no control authority. It has no data saving option, because it records nothing in Synnax.

Create a Kafka write task

In the devices toolbar, right-click the cluster and select Create write task. You can also run the “Create Kafka write task” command.
Enter the Topic, such as synnax-samples.
Click Add channel, then select a channel in Channel.
Set the JSON type of the value: Number, String, or Boolean.
Click (Start).

The task starts to send records at once. Each record holds the value, the channel name, and the timestamp, as in record shape.

from synnax import kafka

cluster = client.devices.retrieve(name="Plant cluster")
temperature = client.channels.retrieve("oven_temperature")

task = client.tasks.create(
    name="Oven to Kafka",
    type="kafka_write",
    rack=cluster.rack,
    config=kafka.WriteConfig(
        device=cluster.key,
        topic="synnax-samples",
        record=kafka.Record(
            channel_pointer="/channel", timestamp_pointer="/timestamp"
        ),
        channels=[kafka.WriteChannel(channel=temperature.key)],
    ),
)
task.start()

Pass rack=cluster.rack. With no rack, Python picks an embedded rack, which can be the rack of the Driver instead of the rack of the Core.

import { kafka } from "@synnaxlabs/client";

const [cluster] = await client.devices.retrieve({
  where: { name: { eq: "Plant cluster" } },
});
const temperature = await client.channels.retrieve("oven_temperature");

const task = await client.tasks.create({
  name: "Oven to Kafka",
  rack: cluster.rack,
  type: "kafka_write",
  config: kafka.writeConfigZ.parse({
    device: cluster.key,
    topic: "synnax-samples",
    record: { channelPointer: "/channel", timestampPointer: "/timestamp" },
    channels: [{ channel: temperature.key }],
  }),
});
await task.start();

Record shape

Each record is one JSON object. Three pointers set where each part goes:

Value pointer -> the sample value. /value by default.
Channel pointer -> the name of the channel. The Console sets /channel. Clear it to leave the name out.
Timestamp pointer -> the timestamp of the sample, in the Time format. The Console sets /timestamp. Clear it to leave the timestamp out.

With the Console defaults, a sample of oven_temperature gives this record value:

{ "value": 182.5, "channel": "oven_temperature", "timestamp": 1767225600000000000 }

A pointer can go deeper than one level, such as /data/value. The record also has a Kafka timestamp, which is the timestamp of the sample.

The timestamp of a sample comes from the index channel of its channel. A virtual channel has no index, so its samples take the time at which the Core receives them.

Record key sets the Kafka key of each record. Channel name puts each channel on its own key, so Kafka keeps the samples of one channel in order. None sends records with no key.

In code, record.fields adds the same fields to every record. A static field adds a fixed value, and a generated field adds a new UUID or the current time. The Console does not show them.

from synnax import http, kafka

kafka.Record(
    channel_pointer="/channel",
    fields=[
        http.StaticWriteField(pointer="/site", json_type="string", value="plant-1"),
        http.GeneratedWriteField(pointer="/id", generator="uuid"),
    ],
)

The field types come from synnax.http. A generated timestamp field takes a time_format, and its default is iso8601.

Task options

Setting	Default	Notes
Topic `topic`		One topic for each task.
Record key `record_key`	Channel name	Channel name or None.
Value pointer `record.value_pointer`	`/value`	Empty uses `/value`.
Channel pointer `record.channel_pointer`	`/channel`	Empty leaves the name out. In code, leave it unset: `""` fails.
Timestamp pointer `record.timestamp_pointer`	`/timestamp`	Empty leaves it out. In code, leave it unset: `""` fails.
Time format `record.time_format`	Unix (ns)	ISO 8601, or Unix time in s, ms, µs, or ns.
Auto start `auto_start`	Off	Starts the task each time the Core starts.

TypeScript uses the same fields in camel case, such as recordKey.

Channel reference

Setting	Default	Notes
Channel `channel`		The channel whose samples the task sends.
JSON type `json_type`	Number	Number, String, or Boolean.
Time format `time_format`	None	Required for a `timestamp` channel. Replaces the JSON type.
Enum mapping `enum_values`	None	With String, maps numbers to text labels.

A string, JSON, or UUID channel always goes out as a string.

When the cluster is unreachable

The task does not start when the Core cannot reach the cluster. While the task runs, a record that the cluster does not accept shows Failed to produce record, and the task keeps running.

Troubleshooting

Failed to produce record
Cause
The task could not send a record. The description gives the reason:
- Produce failed -> the cluster did not accept the record, for example because the user cannot write to the topic.
- Channel … arrived without its index -> a sample came with no timestamp from its index channel.
- Record build failed -> a value does not fit the JSON type or the enum mapping.
The task keeps running.
Fix
- Produce failed -> check the topic name and the ACLs of the user.
- Channel … arrived without its index -> write the index channel in the same frame as the channel.
- Record build failed -> match the JSON type and the enum mapping to the values.
connecting to cluster: ...
Cause
The task did not start, because the Core cannot reach the cluster.
Fix
Check the cluster status, then start the task again.
channels: at least one must be enabled
Cause
The task has no enabled channels.
Fix
Add a channel, or enable one.
channel ...: time_format is required for timestamp channel ...
Cause
A timestamp channel has no time format.
Fix
Set Time format on the channel.

For problems with the connection, see Connect a Kafka cluster.

Read task

Next in LabJack

Configure a device