Comment on page

Kafka

Kafka publisher transport emits processed events onto a defined cluster topic. Events are serialised using either the default Json StreamEvents serializer or a custom transformer implementation.
Driver details
org.apache.kafka:kafka-clients:2.7.0
Example configuration
kafkaPublisher:
 cluster address: localhost:9092
 topic: customers
 partitionKeys:
    - customerId
This simple example leverages the default setting resulting in events published as StreamEvent Json object on to the customers topic using the customer Id as the partitioning key.
An example use case for these settings is when there is a need to chain multiple Joule processors to form a complex, scalable and resilient use case.
Core Attributes
Configuration parameters available for the InfluxDB publisher transport. The parameters are organized by order of importance, ranked from high to low.
Attribute
Description
Data Type
Required
cluster address 
InfluxDB server address
http://<ip-address>:port
topic
InfluxDB UI / CLI Authentication access token
String
partitionKeys
Authentication access details for v1.x InfluxDB. 
Note: Only required if authToken has not been provided.
String
serializer
InfluxDB UI / CLI organisation token 
String
batchSize
Number of events to batch send, maps to batch.size. Batch size of zero disables batching function
Integer
Default: 1024
memBufferSize
Size in bytes of event buffer. Maps to buffer.memory
Long 
Default: 33554432
retries
Maps to retries
Integer 
Default: 0
properties
Additional publisher properties to be applied to the Kafka publisher subsystem
Properties map
Serialization Attributes
This topic provides configuration parameters available for the formatter attribute. 
Attribute
Description
Data Type
Required
transform
User provided implementation
Implementation of CustomTransformer
key serializer
Domain class that maps to the partition key type. Property maps to key.serializer property
String 
Default: IntegerSerializer
value serializer
Domain class that serializes to Kafka. Property maps to value.serializer property
String
Default: StreamEventJsonSerializer
Serializer example
kafkaPublisher:
 cluster address: localhost:9092
 topic: customers
 partitionKeys:
   - customerId
​
 serializer:
   key serializer: org.apache.kafka.common.serialization.IntegerSerializer
   value serializer: com.fractalworks.streams.transport.kafka.serializers.json.StreamEventJsonSerializer
OOTB serializers
A flexible event serialization model is provided to enable Joule processes to be chained for complex use cases and provide down stream integrations. Out of the box Joule serializers are provided, along with Apache Kafka implementation, while custom serialization is achieved through the Joule SDK.
com.fractalworks.streams.transport.kafka.serializers.object.ObjectSerializer
com.fractalworks.streams.transport.kafka.serializers.json.StreamEventJsonSerializer
com.fractalworks.streams.transport.kafka.serializers.avro.AvroStreamEventSerializer
Custom Transformers
By implementing a CustomTransformer you can provide domain specific data types to ease integration to downstream applications.
Example
kafkaPublisher:
 cluster address: localhost:9092
 topic: customers
 partitionKeys:
   - customerId
​
 serializer:
   transform: com.fractalworks.streams.transport.kafka.CustomerTransformer
   key serializer: org.apache.kafka.common.serialization.IntegerSerializer
   value serializer: com.fractalworks.streams.transport.kafka.serializers.object.ObjectSerializer
Java transformer implementation 
This class will convert an internal StreamEvent object to a user defined Customer data type 
public class CustomerTransformer implements CustomTransformer<Customer> {
​
   public CustomerTransformer() {
       // Required
   }
​
   @Override
   public Collection<Customer> transform(Collection<StreamEvent> events) 
   throws TranslationException {
​
       Collection<Customer> customers = new ArrayList<>();
       for (StreamEvent event : events) {
           customers.add(transform(event));
       }
       return customers;
   }
​
   @Override
   public Customer transform(StreamEvent event) 
   throws TranslationException {
​
       Customer customer = new Customer();
       customer.setCustomerId( (int)event.getValue("customerId"));
       customer.setFirstname( (String) event.getValue("firstname"));
       customer.setSurname( (String) event.getValue("surname"));
       return customer;
   }
}

Attribute	Description	Data Type
cluster address	InfluxDB server address	http://<ip-address>:port
topic	InfluxDB UI / CLI Authentication access token	String
partitionKeys	Authentication access details for v1.x InfluxDB. Note: Only required if authToken has not been provided.	String
serializer	InfluxDB UI / CLI organisation token	String
batchSize	Number of events to batch send, maps to batch.size. Batch size of zero disables batching function	Integer Default: 1024
memBufferSize	Size in bytes of event buffer. Maps to buffer.memory	Long Default: 33554432
retries	Maps to retries	Integer Default: 0
properties	Additional publisher properties to be applied to the Kafka publisher subsystem	Properties map

Attribute	Description	Data Type
transform	User provided implementation	Implementation of CustomTransformer
key serializer	Domain class that maps to the partition key type. Property maps to key.serializer property	String Default: IntegerSerializer
value serializer	Domain class that serializes to Kafka. Property maps to value.serializer property	String Default: StreamEventJsonSerializer

Connector Catalog - Previous

Sinks

MQTT

Last modified 1mo ago