Kafka Streams wrapper for Kafka Metamorphosis.
Kafka Streams wrapper for Kafka Metamorphosis.
(aggregate grouped-stream init-fn agg-fn)(aggregate grouped-stream init-fn agg-fn store-name)Aggregate values per key in a grouped (or windowed) stream. init-fn is a no-arg function producing the initial aggregate value. agg-fn is a function of [key value aggregate] -> new-aggregate. Returns a KTable.
Aggregate values per key in a grouped (or windowed) stream. init-fn is a no-arg function producing the initial aggregate value. agg-fn is a function of [key value aggregate] -> new-aggregate. Returns a KTable.
(branch stream predicates)Split a KStream into multiple streams using a sequence of predicate functions. Returns a vector of KStreams in the same order as predicates.
Split a KStream into multiple streams using a sequence of predicate functions. Returns a vector of KStreams in the same order as predicates.
(build-topology builder)Build a Topology from a StreamsBuilder.
Build a Topology from a StreamsBuilder.
(clean-up! streams)Clean up local state for the Kafka Streams instance.
Clean up local state for the Kafka Streams instance.
(close! streams)(close! streams timeout-ms)Close the Kafka Streams instance. Optionally provide a timeout in milliseconds.
Close the Kafka Streams instance. Optionally provide a timeout in milliseconds.
(count-values grouped-stream)(count-values grouped-stream store-name)Count records per key in a grouped (or windowed) stream, returning a KTable.
Count records per key in a grouped (or windowed) stream, returning a KTable.
(create topology config)Create a KafkaStreams instance from an existing Topology and config map.
Create a KafkaStreams instance from an existing Topology and config map.
(create-builder)Create a new Kafka Streams builder.
Create a new Kafka Streams builder.
(create-from-builder builder config)Create a KafkaStreams instance from a StreamsBuilder and config map.
Create a KafkaStreams instance from a StreamsBuilder and config map.
Default Kafka Streams application id.
Default Kafka Streams application id.
(filter-stream stream pred)Filter records from a KStream using a predicate function of [key value].
Filter records from a KStream using a predicate function of [key value].
(flat-map-kv stream f)Transform each [key value] pair of a KStream into zero or more [key value] pairs using f, a function of [key value] that returns a sequence of pairs.
Transform each [key value] pair of a KStream into zero or more [key value] pairs using f, a function of [key value] that returns a sequence of pairs.
(global-table builder topic)(global-table builder topic {:keys [store]})Create a GlobalKTable from a builder and a topic.
Pass a :store name to make the underlying state store queryable via query.
Create a GlobalKTable from a builder and a topic. Pass a :store name to make the underlying state store queryable via `query`.
(group stream)Group a KStream by its existing key, returning a KGroupedStream.
Group a KStream by its existing key, returning a KGroupedStream.
(group-by-key stream f)Group a KStream by a new key computed from [key value] using f, returning a KGroupedStream.
Group a KStream by a new key computed from [key value] using f, returning a KGroupedStream.
(join stream ktable value-joiner-fn)Inner join a KStream with a KTable using value-joiner-fn, a function of [stream-value table-value] -> joined-value.
Inner join a KStream with a KTable using value-joiner-fn, a function of [stream-value table-value] -> joined-value.
(left-join stream ktable value-joiner-fn)Left join a KStream with a KTable using value-joiner-fn, a function of [stream-value table-value] -> joined-value. table-value is nil when there is no match.
Left join a KStream with a KTable using value-joiner-fn, a function of [stream-value table-value] -> joined-value. table-value is nil when there is no match.
(map-kv stream f)Transform both key and value of a KStream using f, a function of [key value] that returns a [new-key new-value] pair.
Transform both key and value of a KStream using f, a function of [key value] that returns a [new-key new-value] pair.
(map-values stream f)Transform the values of a KStream using f.
f receives the current value and should return the transformed value.
Transform the values of a KStream using f. f receives the current value and should return the transformed value.
(query streams store-name key)Query a key-value state store by key from a running KafkaStreams instance.
store-name must match a :store name passed to table, global-table,
count-values, reduce-values, or aggregate.
Query a key-value state store by key from a running KafkaStreams instance. store-name must match a :store name passed to `table`, `global-table`, `count-values`, `reduce-values`, or `aggregate`.
(reduce-values grouped-stream f)(reduce-values grouped-stream f store-name)Reduce values per key in a grouped (or windowed) stream using f, a function of [aggregate value] -> new-aggregate. Returns a KTable.
Reduce values per key in a grouped (or windowed) stream using f, a function of [aggregate value] -> new-aggregate. Returns a KTable.
(start! streams)Start the Kafka Streams instance.
Start the Kafka Streams instance.
(state streams)Return the current Kafka Streams state.
Return the current Kafka Streams state.
(stop! streams)(stop! streams timeout-ms)Stop the Kafka Streams instance. Alias for close!.
Stop the Kafka Streams instance. Alias for `close!`.
(stream builder topics)Create a KStream from a builder and a topic or list of topics.
Create a KStream from a builder and a topic or list of topics.
(streams-config)(streams-config application-id)(streams-config application-id overrides)Create a Kafka Streams configuration map with sensible defaults.
Usage: (streams-config) ; local brokers and default application id (streams-config "my-app") ; custom application id (streams-config "my-app" {:bootstrap-servers "kafka:9092"})
Create a Kafka Streams configuration map with sensible defaults.
Usage:
(streams-config) ; local brokers and default application id
(streams-config "my-app") ; custom application id
(streams-config "my-app" {:bootstrap-servers "kafka:9092"})(table builder topic)(table builder topic {:keys [store]})Create a KTable from a builder and a topic.
Pass a :store name to make the underlying state store queryable via query.
Create a KTable from a builder and a topic. Pass a :store name to make the underlying state store queryable via `query`.
(to stream topic)Send a KStream to the given topic and return the same stream.
Send a KStream to the given topic and return the same stream.
(window grouped-stream size-ms)Apply a tumbling time window of the given size (in milliseconds) to a KGroupedStream, returning a TimeWindowedKStream.
Apply a tumbling time window of the given size (in milliseconds) to a KGroupedStream, returning a TimeWindowedKStream.
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |