MooseStack Documentation

Kafka

Use the Kafka engine to consume data directly from Kafka compatible topics.

For more details on the Kafka engine, see the ClickHouse documentation on Kafka integration.

Kafka with Materialized Views

Kafka tables are streaming interfaces that don't persist data. Use a MaterializedView to continuously move data to a table.

KafkaTable.ts

import { OlapTable, ClickHouseEngines, MaterializedView, sql } from '@514labs/moose-lib'; interface KafkaEvent {  eventId: string;  userId: string;  timestamp: number; // Unix seconds for JSONEachRow} // Kafka table - reads from topic, doesn't persistexport const kafkaSource = new OlapTable<KafkaEvent>("kafka_events", {  engine: ClickHouseEngines.Kafka,  brokerList: "redpanda:9092",  topicList: "events",  groupName: "my_consumer_group",  format: "JSONEachRow",  settings: {    kafka_num_consumers: "1",  }}); // MaterializedView moves data to MergeTree for persistenceconst cols = kafkaSource.columns;export const eventsMV = new MaterializedView<KafkaEvent>({  tableName: "events_dest",  materializedViewName: "events_mv",  orderByFields: ["eventId"],  selectStatement: sql`SELECT ${cols.eventId}, ${cols.userId}, ${cols.timestamp} FROM ${kafkaSource}`,  selectTables: [kafkaSource],});

Kafka Engine Limitations

No ORDER BY: Kafka is a streaming engine and doesn't support orderByFields
No ALTER TABLE: Column or settings changes require DROP and CREATE (Moose handles this automatically)
No direct SELECT: Query data from the MaterializedView destination table, not the Kafka table

import { OlapTable, ClickHouseEngines, MaterializedView, sql } from '@514labs/moose-lib'; interface KafkaEvent { eventId: string; userId: string; timestamp: number; // Unix seconds for JSONEachRow} // Kafka table - reads from topic, doesn't persistexport const kafkaSource = new OlapTable<KafkaEvent>("kafka_events", { engine: ClickHouseEngines.Kafka, brokerList: "redpanda:9092", topicList: "events", groupName: "my_consumer_group", format: "JSONEachRow", settings: { kafka_num_consumers: "1", }}); // MaterializedView moves data to MergeTree for persistenceconst cols = kafkaSource.columns;export const eventsMV = new MaterializedView<KafkaEvent>({ tableName: "events_dest", materializedViewName: "events_mv", orderByFields: ["eventId"], selectStatement: sql`SELECT ${cols.eventId}, ${cols.userId}, ${cols.timestamp} FROM ${kafkaSource}`, selectTables: [kafkaSource],});