Skip to main content

End-to-End Example: Setting up a Striim Application Using Kafka Writer

To build a complete pipeline, work through the following sections of this guide in order:This section walks through a complete, runnable pipeline: a database source mapped explicitly to per-table Kafka topics, formatted as Avro, and registered against a Confluent Schema Registry. To build a complete pipeline for your own scenario, work through the following sections of this guide in order:

  1. Create the Kafka Connection Profile — see Kafka Writer Initial Setup, including the authentication method for your cluster (SSL-only, SASL, or Mutual TLS) and, for Amazon MSK, Writing to AWS MSK.

  2. Store certificates and credentials in the Striim Vault, if your authentication method requires them — see Using the Striim Vault.

  3. Create the Schema Registry Connection Profile, if using the Avro Formatter — see Connecting to Schema Registry.

  4. Create the Striim application (source, optional CQ, and KafkaWriter target) — see Create Kafka Writer Application and the Sample TQLs for complete, runnable examples.

  5. Deploy, run, and monitor the application, and verify messages are arriving in the target Kafka topics — see Kafka Writer Operational Considerations for the available metrics.

Worked Example: Database Source to Kafka with Explicit Topic Mapping, Avro, and Schema Registry

This application reads three tables from a source database and writes each to its own explicitly named Kafka topic, using Avro formatting with a Confluent Schema Registry connection profile and exactly-once processing (E1P).

CREATE OR REPLACE APPLICATION DBToKafka;

CREATE OR REPLACE SOURCE DB_Source USING Global.DatabaseReader (
  Tables: 'waction.Bank;waction.Person;waction.retailData;',
  DatabaseProviderType: 'Default',
  FetchSize: 100,
  ConnectionURL: 'jdbc:mysql://<DB_HOST>:3306/waction',
  Username: '<DB_USERNAME>',
  Password_encrypted: 'true',
  Password: '<ENCRYPTED_PASSWORD>' )
OUTPUT TO DB_Data;

CREATE OR REPLACE TARGET KTWExplicitTopic USING Global.KafkaWriter (
  ConnectionProfileName: 'admin.KafkaCp',
  Topics: 'waction.Bank,Bank;waction.Person,Person;waction.retailData,retailData',
  TopicKey: '@MetaData(TableName)',
  MessageKey: 'Custom',
  CustomMessageKey: 'OperationName=@MetaData(OperationName)',
  MessageHeader: 'CompanyName="Striim"',
  PartitionKey: 'None',
  AutoCreateTopic: true,
  DataTopicConfig: '{"PartitionCount":"1","ReplicationFactor":"3"}',
  CheckpointTopicConfig: '{"PartitionCount":1,"ReplicationFactor":3,"CleanUpPolicy":"compact"}',
  E1P: true,
  Serializer: 'StriimSerializer',
  SchemaEvolution: 'Auto',
  CommitPolicy: 'EventCount=10000;Interval=15s',
  ConnectionRetryPolicy: 'initialRetryDelay=10s,retryDelayMultiplier=2,maxRetryDelay=1m,maxAttempts=5,totalTimeout=10m' )
FORMAT USING Global.AvroFormatter (
  useSchemaRegistryConnectionProfile: 'true',
  SchemaRegistryConnectionProfileName: 'admin.confluentSchemaRegistryCP',
  formatAs: 'Native',
  SchemaRegistrySubjectName: 'UseTopicName' )
INPUT FROM DB_Data;

END APPLICATION DBToKafka;

What this application does, property by property:

Topics maps each source table to a distinct, explicitly named Kafka topic (source-table,topic-name pairs, semicolon-separated). TopicKey selects the table name from event metadata to drive that mapping.

ConnectionProfileName and SchemaRegistryConnectionProfileName reference the Kafka and Confluent Schema Registry connection profiles created in Steps 1 and 3 above — see Kafka Writer Initial Setup and Connecting to Schema Registry for how to create them.

E1P: true enables exactly-once processing; AutoCreateTopic and CheckpointTopicConfig let Kafka Writer create the data and checkpoint topics automatically rather than requiring them to be pre-created.

formatAs: 'Native' with SchemaRegistrySubjectName: 'UseTopicName' registers each table's schema under a subject named for its Kafka topic — see Avro Formatter and Subject Name Mapping for the other available options.

Replace the placeholders in angle brackets (<DB_HOST>, <DB_USERNAME>, <ENCRYPTED_PASSWORD>) with values for your own database before deploying.