Kafka Writer

request.timeout.ms=60001;
session.timeout.ms=60000

MessageKey : CName=”Striim”

MessageKey : Table=@metadata(TableName);
  Operation=@metadata(OperationName);key1=@userdata(key1)

MessageKey : CityName=City; Zipcode=zip

MessageKey : CName=”Striim”;Table=@metadata(TableName);
  Operation=@metadata(OperationName)

This adapter has a choice of formatters. See Supported writer-formatter combinations for more information.Supported writer-formatter combinations

With the exceptions noted in the following table, you may specify any Kafka producer property in KafkaConfig. 

Internally, KafkaWriter invokes KafkaConsumer for various purposes, and the WARNING from the consumer API due to passing KafkaConfig  properties can be safely ignored.

The following sample code writes data from PosDataPreview.csv to the Kafka topic KafkaWriterSample. This topic already exists in Striim's internal Kafka instance. If you are using an external Kafka instance, you must create the topic before running the application.

CREATE SOURCE PosSource USING FileReader (
  directory:'Samples/PosApp/AppData',
  wildcard:'PosDataPreview.csv',
  positionByEOF:false
)
PARSE USING DSVParser (
  header:yes
)
OUTPUT TO RawStream;

CREATE CQ CsvToPosData
INSERT INTO PosDataStream
SELECT TO_STRING(data[1]) as merchantId,
  TO_DATEF(data[4],'yyyyMMddHHmmss') as dateTime,
  TO_DOUBLE(data[7]) as amount,
  TO_STRING(data[9]) as zip
FROM RawStream;

CREATE TARGET KW11Sample USING KafkaWriter VERSION '2.1.0'(
  brokeraddress:'localhost:9092',
  topic:'KafkaWriterSample'
)
FORMAT USING DSVFormatter ()
INPUT FROM PosDataStream;

You can verify that data was written to Kafka by running the Kafka Reader sample application.

The first field in the output (position) stores information required to avoid lost or duplicate events after recovery (see Recovering applications). If recovery is not enabled, its value is NULL.

mon output (see Using the MON command) for targets using KafkaWriter includes:

When you enable compression in KafkaWriter, the broker and consumer should handle the compressed batches automatically. No additional configuration should be required in Kafka.

To enable compression for version 0.11, include the compression.type property in KafkaConfig. Supported values are gzip, lz4, snappy. For example:

KafkaConfig:'compression.type=snappy'

When using Kafka Writer to write to Confluent Cloud (Kafka as a service), additional configuration properties are required. In particular, you must set ssl.endpoint.identification.algorithm==https in the KafkaConfig string, which is required for Confluent Cloud clusters using secure endpoints. This property is not needed for self-hosted Kafka or other cloud providers.

The full KafkaConfig string typically looks like:

KafkaConfig:
'session.timeout.ms==60000:
  sasl.mechanism==PLAIN:
  sasl.jaas.config==org.apache.kafka.common.security.plain.PlainLoginModule required
    username="CLUSTER_API_KEY"
    password="CLUSTER_API_SECRET";
  ssl.endpoint.identification.algorithm==https:
  security.protocol==SASL_SSL'

You can generate the required credentials and configuration strings using the Confluent Cloud UI under Clients > Java, or by using the Confluent Cloud CLI.

If the INPUT FROM stream is partitioned, events will be distributed among Kafka partitions based on the values in the input stream's PARTITION BY property. All events with the same value in the PARTITION BY field will be written to the same randomly selected partition. Striim will distribute the data evenly among the partitions to the extent allowed by the frequency of the various PARTITION BY field values (for example, if 80% of the events have the same value, then one of the Kafka partitions will contain at least 80% of the events). In the example above, to enable partitioning by city, you would revise the definition of TransformedDataStream as follows:

CREATE STREAM TransformedDataStream OF TransformedDataType PARTITION BY City;

To override this default behavior and send events to specific partitions based on the PARTITION BY field values, see Creating a custom Kafka partitioner.

KafkaWriter performs differently depending on whether the mode property value is sync or async and whether recovery (see Recovering applications) is enabled for the application. The four possibilities are:

When using sync, multiple events are batched in a single Kafka message. The number of messages in a batch is controlled by the batch.size parameter, which by default is 1 million bytes. The maximum amount of time KafkaWriter will wait between messages is set by the linger.ms parameter, which by default is 1000 milliseconds. Thus, by default, KafkaWriter will write a message after it has received a million bytes or one second has elapsed since the last message was written, whichever occurs first.

 Batch.size must be larger than the largest event KafkaWriter will receive, but must not exceed the max.message.bytes size in the Kafka topic configuration.

The following setting would write a message every time KafkaWriter has received 500,000 bytes or two seconds has elapsed since the last message was written:

KafkaConfig:'batch.size=500000,linger.ms=2000'

Each output record will begin with a field containing information Striim can use to ensure that no duplicate records are written during recovery (see Recovering applications).

For input events of this user-defined Striim type:

CREATE TYPE emptype ( id int, name string);

output would be similar to:

1234002350,1,User One

For input events of type WAEvent, output (Position, TableName, CommitTimestamp, Operation, data) would be similar to:

SCN:1234002344,Employee,12-Dec-2016 19:13:00,INSERT,1,User One

Each JSON output node will contain a field named __striimmetadata with a nested field position containing information Striim can use to ensure that no duplicate records are written during recovery (see Recovering applications).

For input events of this user-defined Striim type:

CREATE TYPE emptype ( id int, name string);

output would be similar to:

{
    "ID": 1,
    "Name": "User One", 
    "__striimmetadata"  :  {"position" : "SCN:1234002344" } 
}

For input events of type WAEvent, output would be similar to:

{
"metadata" : { "TABLENAME" : "Employee","CommitTimestamp" : "12-Dec-2016 19:13:00", 
  "OperationName" : "INSERT"  }
"data" : { 
      "ID" : "1",
      "NAME" : "User One"},
"__striimmetadata" :  { "position" : "SCN:1234002350" }   // but in binary format
}

Each output record will include an extra element __striimmetadata with a nested element position containing information Striim can use to ensure that no duplicate records are written during recovery (see Recovering applications).

For input events of this user-defined Striim type:

CREATE TYPE emptype ( id int, name string);

output would be similar to:

...
<ID> 1 </ID>
<Name> User One </Name> 
<__striimmetadata>
  <position>1234002344</position>
</__striimmetadata> ...