Retrieve Kafka Properties

Retrieving Kafka Properties from a JSON Pipeline

The following CREATE TABLE statement creates a new table t, and adds four properties to each Kafka message. If the property names are prop_offset, prop_partition, prop_timestamp, and prop_topicname, then:

1. Create a new table t.

   SQL

   Copy

   CREATE TABLE t(a INT, b JSON, c INT, offset Text, partition Text, timestamp Text, topicname Text);

2. Create a pipeline to load data into this new table.

   SQL

   Copy

   CREATE OR REPLACE PIPELINE p AS
   LOAD DATA kafka 'localhost:9092/jsontopic'
   INTO TABLE t FORMAT JSON (a <- json_a, b <- json_b, c <- json_c)
   SET offset = get_kafka_pipeline_prop("prop_offset"),
   partition = get_kafka_pipeline_prop("prop_partition"),
   timestamp = get_kafka_pipeline_prop("prop_timestamp"),
   topicname = get_kafka_pipeline_prop("prop_topicname");

Example: Retrieving Kafka Properties with a Stored Procedure

The following example shows how to retrieve Kafka properties with a Stored Procedure. A CONVERT clause is also used to typecast the properties into usable data types. For more information on Stored Procedures, refer to CREATE PIPELINE ... INTO PROCEDURE.

1. Create a new table t.

   SQL

   Copy

   CREATE TABLE t(a INT, b INT, c INT,
   offset BIGINT, partition BIGINT, timestamp BIGINT, topicname Text);

2. Create a Stored Procedure with the CONVERT clause to typecast the Kafka properties.

   SQL

   Copy

   CREATE OR REPLACE PROCEDURE test_proc(
   batch QUERY( a INT, b INT, c INT,
   offset TEXT, partition TEXT, timestamp TEXT, topicname TEXT)
   AS BEGIN INSERT INTO t(a, b, c, offset, partition, timestamp, topicname)
   SELECT a, b, c,
   CONVERT(offset,SIGNED),
   CONVERT(partition, SIGNED),
   CONVERT(timestamp, SIGNED),
   topicname
   FROM batch;

3. Run a pipeline and retrieve the Kafka properties with the Stored Procedure.

   SQL

   Copy

   CREATE OR REPLACE PIPELINE p
   AS LOAD DATA kafka 'localhost:9092/jsontopic'
   INTO PROCEDURE test_proc FORMAT JSON
   (a <- json_a, b <- json_b, c <- json_c)
   set offset = get_kafka_pipeline_prop("prop_offset"),
   partition = get_kafka_pipeline_prop("prop_partition"),
   timestamp = get_kafka_pipeline_prop("prop_timestamp"),
   topicname = get_kafka_pipeline_prop("prop_topicname");

4. Query the table t to view the typecasted properties.

   SQL

   Copy

   SELECT * FROM t;

   +----+----+----+------+---------+-------------+---------+
   | a | b | c | offset | partition | timestamp | topicname |
   +----+----+----+------+---------+-------------+---------+
   | 4 | 5 | 6 | 1 | 0 | 1738318906369 | jsontopic |
   | 1 | 2 | 3 | 0 | 0 | 1738317123940 | jsontopic |
   +----+----+----+------+---------+-------------+---------+

Example: Retrieving Kafka Message Key

A Kafka message includes both a payload and a message key. Use the KAFKA KEY clause to extract the message key and store it in a table column as part of pipeline ingestion. SingleStore supports Kafka message keys only for Avro and JSON payloads, and you must encode the key in the corresponding Avro and JSON format.

The following example creates a table and a pipeline that extracts the Kafka message key and Kafka message properties.

1. Create a new table t.

   SQL

   Copy

   CREATE TABLE t (
       inp_json JSON,
       kafka_key TEXT,
       kafka_partition INT,
       kafka_offset BIGINT,
       kafka_timestamp BIGINT
   );

2. Create and run a pipeline to ingest Kafka messages into the table t, including the message payload, message key, and Kafka metadata.

   SQL

   Copy

   CREATE OR REPLACE PIPELINE p AS
   LOAD DATA KAFKA 'localhost:9092/test_topic'
   INTO TABLE t
   FORMAT JSON (inp_json <- %)
   KAFKA KEY (kafka_key <- %)
   SET kafka_partition = get_kafka_pipeline_prop("prop_partition"),
       kafka_offset = get_kafka_pipeline_prop("prop_offset"),
       kafka_timestamp = get_kafka_pipeline_prop("prop_timestamp");

Kafka Headers

Kafka headers are metadata key-value pairs that you can add to Kafka messages. They provide additional context or information about the message without affecting the message payload. Each header consists of a key and a value. 

The get_kafka_pipeline_prop("propname") function can be used to consume these headers. For example, if a Kafka message has the headers:

message_headers = [
(“Country”,”USA”),
(“City”, “Arizona”),
(“Language”, “English”),
]

Enter the header_key (“Country”, “City”, and “Language” in this example) in place of the “propname” in the argument of the get_kafka_pipeline_prop("propname") function to ingest the headers. 

For example:

1. Create a new table t.

   SQL

   Copy

   CREATE TABLE t (info text, country text, city text, language text);

2. Create a pipeline to load data into this new table.

   SQL

   Copy

   CREATE PIPELINE p AS
   LOAD DATA KAFKA 'localhost:9092/topic_name'
   (info)
   SET country = get_kafka_pipeline_prop("Country"),
   city = get_kafka_pipeline_prop("City"),
   language = get_kafka_pipeline_prop("Language");