org.apache.kafka.streams.kstream

Interface Transformer<K,V,R>

Type Parameters:

K - key type

V - value type

R - KeyValue return type (both key and value type can be set arbitrarily)

@InterfaceStability.Unstable
public interface Transformer<K,V,R>

The Transformer interface for stateful mapping of an input record to zero, one, or multiple new output records (both key and value type can be altered arbitrarily). This is a stateful record-by-record operation, i.e, transform(Object, Object) is invoked individually for each record of a stream and can access and modify a state that is available beyond a single call of transform(Object, Object) (cf. KeyValueMapper for stateless record transformation). Additionally, the interface can be called in regular intervals based on the processing progress (cf. punctuate(long).

Use TransformerSupplier to provide new instances of Transformer to Kafka Stream's runtime.

If only a record's value should be modified ValueTransformer can be used.

See Also:

TransformerSupplier, KStream.transform(TransformerSupplier, String...), ValueTransformer, KStream.map(KeyValueMapper), KStream.flatMap(KeyValueMapper)

Method Summary

Methods

Modifier and Type	Method and Description
void	close() Close this processor and clean up any resources.
void	init(ProcessorContext context) Initialize this transformer.
R	punctuate(long timestamp) Perform any periodic operations and possibly generate new KeyValue pairs if this processor schedules itself with the context during initialization.
R	transform(K key, V value) Transform the record with the given key and value.

Method Detail

init

void init(ProcessorContext context)

Initialize this transformer. This is called once per instance when the topology gets initialized.

The provided context can be used to access topology and record meta data, to schedule itself for periodical calls (cf. punctuate(long)), and to access attached StateStores.

Note, that ProcessorContext is updated in the background with the current record's meta data. Thus, it only contains valid record meta data when accessed within transform(Object, Object).

Parameters:

context - the context

transform

R transform(K key,
          V value)

Transform the record with the given key and value. Additionally, any state that is attached to this operator can be accessed and modified arbitrarily (cf. ProcessorContext.getStateStore(String)).

If more than one output record should be forwarded downstream ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), and ProcessorContext.forward(Object, Object, String) can be used. If not record should be forwarded downstream, transform can return null.

Parameters:

key - the key for the record

value - the value for the record

Returns:

new KeyValue pair—if null no key-value pair will be forwarded to down stream

punctuate

R punctuate(long timestamp)

Perform any periodic operations and possibly generate new KeyValue pairs if this processor schedules itself with the context during initialization.

To generate new KeyValue pairs ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), and ProcessorContext.forward(Object, Object, String) can be used.

Note that punctuate is called based on stream time (i.e., time progresses with regard to timestamps return by the used TimestampExtractor) and not based on wall-clock time.

Parameters:

timestamp - the stream time when punctuate is being called

Returns:

must return null—otherwise, a exception will be thrown

close

void close()

Close this processor and clean up any resources.

To generate new KeyValue pairs ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), and ProcessorContext.forward(Object, Object, String) can be used.