Create a ClickHouse Writer application

With Table Engine set to MergeTree and Mode set to MERGE, ClickHouse Writer reconciles INSERT, UPDATE, DELETE, and PKUPDATE operations from the source into the target table, using the DeleteStrategy property to control how rows are physically removed (LightweightDelete or MutationDelete). For a full row image on UPDATE, the target row is correctly replaced with the new values. For a partial row image (only the changed columns are present), enable OptimizedMerge so that ClickHouse Writer correctly reconstructs the full row, retaining existing values for any column the source did not send. PKUPDATE (a change to the primary key) removes the old key and inserts a row for the new key, correctly carrying over unchanged columns even when the source sends only a partial image.

ClickHouse Writer adds two columns to a ReplacingMergeTree target: a monotonically increasing version column and a soft-delete flag. Currently, Striim manages both columns itself using the fixed names __STRIIM_VERSION and __STRIIM_ISDELETED; custom column names and source-provided version or is-deleted columns are not yet supported. INSERT and UPDATE both write a new row stamped with a higher version; DELETE writes a new row with the delete flag set. Because these are metadata-only writes, ReplacingMergeTree requires only INSERT statements from ClickHouse Writer, which is why it offers higher throughput than MergeTree in MERGE mode.

ClickHouse Writer adds a sign column to a CollapsingMergeTree target. Currently, Striim manages this column itself using the fixed name __STRIIM_SIGN; custom column names and source-provided sign columns are not yet supported. INSERT writes a row with Sign=1. UPDATE writes a cancellation row for the previous state (Sign=-1) followed by a new row for the current state (Sign=1). DELETE writes a single cancellation row (Sign=-1).

With CoalescingMergeTree, ClickHouse Writer writes UPDATE events with only the changed columns populated; every other non-key column is written as NULL, which ClickHouse interprets as "leave this column unchanged." Every non-key column in a CoalescingMergeTree target must be declared Nullable, regardless of whether the source column is nullable. DELETE is handled according to the configured Delete Strategy, the same as with MergeTree. Query with SELECT ... FINAL to force ClickHouse to resolve unmerged parts at read time.

Both engines support INSERT only. If the source emits an UPDATE, DELETE, or PKUPDATE event for a table using either engine, ClickHouse Writer halts the application. Use these engines only for genuinely additive, insert-only sources.

For a Distributed target, ClickHouse Writer resolves the underlying local table on the node it is connected to, and writes to the Distributed table itself using the same INSERT semantics as the underlying local engine; ClickHouse routes each row to the appropriate shard by the configured sharding key. Because Striim does not manage DDL for Distributed tables, both the Distributed table and its underlying local tables must already exist, with matching engine and schema, on every node before you start the application.

Initial schema creation is supported for MergeTree, ReplacingMergeTree, CollapsingMergeTree, CoalescingMergeTree, and SummingMergeTree. It is not supported for AggregatingMergeTree, Replicated table engines, or Distributed tables; for these engines, you must create the target table yourself before deploying the application.

Before creating a table, Striim checks whether a table with that name already exists in ClickHouse. If it does not, Striim creates it. If it does, Striim compares column names and the ORDER BY, PARTITION BY, PRIMARY KEY, and SAMPLE BY clauses; if these are equivalent, Striim treats the table as already correctly created and does not issue a CREATE TABLE statement.

A target column is created NOT NULL when any of the following apply; otherwise it is created nullable:

The NOT NULL override for a sorting-key or partition-key column applies only when Striim can extract a plain column name from the ORDER BY / PARTITION BY expression, or a single-column unary function of one (for example, toYYYYMM(placed_at)). Multi-argument functions, nested functions, arithmetic expressions, and literals are not overridden; the referenced columns instead follow the normal rules above.

Exception — CoalescingMergeTree: every non-sorting-key column is created nullable, regardless of source nullability, because CoalescingMergeTree relies on NULL meaning "leave this column unchanged."

Known limitation: Initial Schema Creation currently creates non-primary-key source columns that are NOT NULL at the source as nullable in the ClickHouse target — only primary-key, sorting-key, and partition-key columns are created NOT NULL. This is temporary; once the source-reader issue is resolved, Initial Schema Creation will produce the same nullability as the CDDL CREATE TABLE flow, which already applies the full rule above.

Examples

MergeTree. The source table's columns are migrated as-is; no extra columns are added.

Source (MySQL):

CREATE TABLE src.employee (
    employee_id INT PRIMARY KEY,
    name        VARCHAR(255) NOT NULL,
    department  VARCHAR(64),
    salary      DECIMAL(10,2),
    hired_at    TIMESTAMP
);

Writer configuration: TableEngine: MergeTree, Tables: 'src.employee, tgt.employee'

Resulting ClickHouse table:

CREATE TABLE tgt.employee (
    employee_id UInt32        NOT NULL,
    name        String        NOT NULL,
    department  String        NULL,
    salary      Decimal(10,2) NULL,
    hired_at    DateTime      NULL
)
ENGINE = MergeTree
ORDER BY employee_id;

employee_id is NOT NULL as the primary key and default sorting key; name is NOT NULL because the source declares it NOT NULL; the rest are nullable since the source has no NOT NULL constraint on them.

ReplacingMergeTree. Adds __STRIIM_VERSION and __STRIIM_ISDELETED.

Source (PostgreSQL):

CREATE TABLE src.orders (
    order_id    INT PRIMARY KEY,
    customer_id INT,
    total       DECIMAL(10,2),
    status      VARCHAR(20),
    placed_at   TIMESTAMP
);

Writer configuration: TableEngine: ReplacingMergeTree, Tables: 'src.orders, tgt.orders'

Resulting ClickHouse table:

CREATE TABLE tgt.orders (
    order_id           UInt32        NOT NULL,
    customer_id        UInt32        NULL,
    total              Decimal(10,2) NULL,
    status             String        NULL,
    placed_at          DateTime      NULL,
    __STRIIM_VERSION   UInt64        NOT NULL,
    __STRIIM_ISDELETED UInt8         NOT NULL
)
ENGINE = ReplacingMergeTree(__STRIIM_VERSION, __STRIIM_ISDELETED)
ORDER BY order_id;

CollapsingMergeTree. Adds __STRIIM_SIGN.

Same source table as above. Writer configuration: TableEngine: CollapsingMergeTree, Tables: 'src.orders, tgt.orders'

Resulting ClickHouse table:

CREATE TABLE tgt.orders (
    order_id      UInt32        NOT NULL,
    customer_id   UInt32        NULL,
    total         Decimal(10,2) NULL,
    status        String        NULL,
    placed_at     DateTime      NULL,
    __STRIIM_SIGN Int8          NOT NULL
)
ENGINE = CollapsingMergeTree(__STRIIM_SIGN)
ORDER BY order_id;

CoalescingMergeTree. No Striim-managed meta columns, but every non-sorting-key column is created NULL regardless of source nullability — this is the one engine where a source NOT NULL constraint is not honored.

Source (MySQL):

CREATE TABLE src.product_inventory (
    product_id   INT PRIMARY KEY,
    name         VARCHAR(100) NOT NULL,
    category     VARCHAR(50),
    price        DECIMAL(10,2),
    stock_qty    INT,
    warehouse_id INT
);

Writer configuration: TableEngine: CoalescingMergeTree, Tables: 'src.product_inventory, tgt.product_inventory'

Resulting ClickHouse table:

CREATE TABLE tgt.product_inventory (
    product_id   UInt32        NOT NULL,
    name         String        NULL,
    category     String        NULL,
    price        Decimal(10,2) NULL,
    stock_qty    Int32         NULL,
    warehouse_id Int32         NULL
)
ENGINE = CoalescingMergeTree()
ORDER BY product_id;

Note: name is nullable here even though the source declares it NOT NULL — specific to CoalescingMergeTree, since a written NULL means "leave this column unchanged."

SummingMergeTree. Append-only, no extra columns; numeric non-key columns are summed on merge for rows sharing the same sorting key.

Source (MySQL):

CREATE TABLE src.daily_sales (
    product_id INT,
    sale_date  DATE,
    revenue    DECIMAL(18,2),
    units_sold INT,
    region     VARCHAR(20),
    PRIMARY KEY (product_id, sale_date)
);

Writer configuration: TableEngine: SummingMergeTree, Tables: 'src.daily_sales, tgt.daily_sales KeyColumn(product_id, sale_date)'

Resulting ClickHouse table:

CREATE TABLE tgt.daily_sales (
    product_id UInt32        NOT NULL,
    sale_date  Date          NOT NULL,
    revenue    Decimal(18,2) NULL,
    units_sold UInt32        NULL,
    region     String        NULL
)
ENGINE = SummingMergeTree()
ORDER BY (product_id, sale_date);

If Striim's comparison finds a mismatch between the Target Table Definition clauses and a pre-existing table's actual ORDER BY, PARTITION BY, PRIMARY KEY, or SAMPLE BY, or a mismatch between the configured Table Engine and the pre-existing table's actual engine family,

To resolve this, either update Target Table Definition or Table Engine in TQL to match the existing table, alter the ClickHouse table to match your TQL configuration, or set RestartOnILBehaviour = replaceTargetTableData on the source so Striim drops and recreates the target table on the next run.

ClickHouse Writer does not load target table metadata when the application starts. Instead, the first time an event for a given target table arrives, ClickHouse Writer fetches that table's metadata (columns, engine, sorting key, and related definition) from ClickHouse and caches it for the life of the application, refreshing it only when a schema change is processed. This means a target table does not need to exist at deploy time — it only needs to exist by the time the first event for it is processed — but it also means a missing or misconfigured table is not reported until that first event arrives, rather than at startup.

When a supported source captures DDL, ClickHouse Writer can respond to it according to the CDDLAction property:

Schema evolution is not supported for AggregatingMergeTree, Replicated table engines, or Distributed tables.

Use the Target Table Definition property to control the DDL clauses Striim uses when it creates a target table. Target Table Definition is a JSON object; each key is a fully qualified target table name, and each value is an object of clauses: Order By, Primary Key, Partition By, Sample By, TTL, and Settings.

Target Table Definition:
{
  "target_db.order_items": {
    "Order By": "order_id",
    "Partition By": "toYYYYMM(created_at)"
  }
}

Order By (sorting key)

Striim resolves the sorting key in this order: an explicit Order By clause in Target Table Definition; otherwise the KeyColumn columns specified in the Tables property; otherwise the source table's primary key columns. If none of these are available, Striim emits no sorting key (tuple()), which Striim reports as a warning and which ClickHouse recommends avoiding for performance reasons. A sorting key is required for ReplacingMergeTree.

Primary Key

An explicit Primary Key clause in Target Table Definition takes priority and is emitted verbatim. If KeyColumn is set together with an explicit Order By clause, the KeyColumn columns become the PRIMARY KEY. If neither is set, Striim emits no PRIMARY KEY clause, and ClickHouse uses the ORDER BY clause as the implicit index.

Partition By

Partition By is optional but recommended for CDC workloads, because without it, updates and deletes scan the entire table. Common patterns include toYYYYMM(<date column>) for monthly partitions (the most common choice for CDC) and toYYYYMMDD(<date column>) for daily partitions on high-volume tables. Avoid partitioning by a high-cardinality column such as a customer or user ID; ClickHouse recommends keeping the total partition count under approximately 1,000.

Sample By, TTL, and Settings

Sample By is supported only on MergeTree-family tables, and its expression must evaluate to an unsigned integer that is part of, or derived from, the table's sorting key. TTL configures automatic row expiry based on a Date or DateTime column. Settings pass ClickHouse table settings through verbatim as comma-separated key=value pairs; quote string values (for example, clean_deleted_rows = 'Always') and leave numeric values unquoted (for example, index_granularity = 8192). An unrecognized key or an incorrectly quoted value causes the CREATE TABLE statement to fail.

If your target table includes ClickHouse columns with DEFAULT, ALIAS, MATERIALIZED, or EPHEMERAL expressions, ClickHouse Writer handles them as follows.

For the different column types in ClickHouse, the different statements are handled as follows:

Note: Initial Schema Creation does not generate DEFAULT, ALIAS, MATERIALIZED, or EPHEMERAL column definitions on a target table it creates, even if the source column has a DEFAULT constraint — these ClickHouse-specific column kinds can only exist on a target table you create or modify yourself. The behavior above applies once such a column exists on the target table, however it was created.

Result

Striim creates the target tables in ClickHouse using the mapped columns, KeyColumn values, and Target Table Definition clauses you configured, and loads the initial data as a batch of INSERT statements. Verify the result by comparing row counts between the source tables and the corresponding ClickHouse tables.

CREATE OR REPLACE APPLICATION OracleToClickhouse_IL RECOVERY 30 SECOND INTERVAL USE EXCEPTIONSTORE TTL : '7d';

CREATE FLOW OracleToClickhouse_IL_SourceFlow;

CREATE OR REPLACE SOURCE OracleToClickhouse_IL_DBSource USING Global.DatabaseReader (
  EnablePartitionAwareRead: true,
  connectionProfileName: 'admin.Oracle_Connection_Profile',
  QuiesceOnILCompletion: true,
  DatabaseProviderType: 'Oracle',
  useConnectionProfile: true,
  Tables: '"<schema>"."<table>"',
  ParallelThreads: 1,
  RestartBehaviourOnILInterruption: 'keepTargetTableData',
  FetchSize: 10000,
  CreateSchema: true )
OUTPUT TO OracleToClickhouse_IL_OutputStream;

END FLOW OracleToClickhouse_IL_SourceFlow;

CREATE OR REPLACE TARGET OracleToClickhouse_IL_Target USING Global.ClickHouseWriter (
  DeleteStrategy: 'LightweightDelete',
  Mode: 'APPENDONLY',
  NullMarker: 'NULL',
  connectionProfileName: 'admin.ClickHouse_Cloud_CP',
  streamingUpload: 'false',
  optimizedMerge: false,
  Tables: '"<schema>"."%","<target_schema>".%',
  UploadPolicy: 'eventcount:10000,interval:30s',
  CDDLAction: 'Process',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=10, totalTimeout=10m',
  TableEngine: 'MergeTree' )
INPUT FROM OracleToClickhouse_IL_OutputStream;

END APPLICATION OracleToClickhouse_IL;

Result

Striim applies ongoing INSERT, UPDATE, and DELETE operations from Oracle to the corresponding ClickHouse tables according to the Table Engine and Mode you configured. Verify the result by making a change in Oracle and confirming it appears in ClickHouse within the interval configured in Upload Policy; for ReplacingMergeTree, query with SELECT ... FINAL to see the deduplicated result.

CREATE OR REPLACE APPLICATION OracleToClickhouse_CDC RECOVERY 30 SECOND INTERVAL USE EXCEPTIONSTORE TTL : '7d';

CREATE FLOW OracleToClickhouse_CDC_SourceFlow;

CREATE OR REPLACE SOURCE OracleToClickhouse_CDC_OracleSource USING Global.OracleReader (
  useConnectionProfile: true,
  connectionProfileName: 'admin.Oracle_Connection_Profile',
  CommittedTransactions: true,
  Tables: '"<schema>"."<table>"',
  TransactionBufferType: 'Memory',
  DatabaseRole: 'PRIMARY',
  connectionRetryPolicy: 'timeOut=30, retryInterval=30, maxRetries=3',
  CDDLAction: 'Process',
  DictionaryMode: 'OnlineCatalog',
  CDDLCapture: false,
  FilterTransactionBoundaries: true,
  SendBeforeImage: true )
OUTPUT TO OracleToClickhouse_CDC_OutputStream;

END FLOW OracleToClickhouse_CDC_SourceFlow;

CREATE OR REPLACE TARGET OracleToClickhouse_CDC_Target USING Global.ClickHouseWriter (
  DeleteStrategy: 'LightweightDelete',
  NullMarker: 'NULL',
  connectionProfileName: 'admin.ClickHouse_Cloud_CP',
  Tables: '"<schema>"."%","<target_schema>".%',
  UploadPolicy: 'eventcount:10000,interval:30s',
  optimizedMerge: true,
  Mode: 'MERGE',
  CDDLAction: 'Process',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=10, totalTimeout=10m',
  TableEngine: 'MergeTree' )
INPUT FROM OracleToClickhouse_CDC_OutputStream;

END APPLICATION OracleToClickhouse_CDC;

Salesforce Reader emits a fully typed WAEvent, so you can map its output directly to ClickHouse Writer without an intermediate typed stream or continuous query. For an initial sync followed by ongoing incremental polling, set Salesforce Reader's Mode to Automated, set Migrate Schema to true so Striim creates the target tables, and set a Polling Interval (for example, 5 minutes; the maximum is 120 minutes). Choose MergeTree with Mode=MERGE for strong consistency, or ReplacingMergeTree for higher throughput on high-change objects.

CREATE OR REPLACE APPLICATION SalesforceToClickhouse RECOVERY 5 MINUTE INTERVAL USE EXCEPTIONSTORE TTL : '7d';

CREATE FLOW SalesforceToClickhouse_SourceFlow;

CREATE OR REPLACE SOURCE SalesforceToClickhouse_Source USING Global.SalesForceReader (
  OAuthAuthorizationFlows: 'PASSWORD',
  autoAuthTokenRenewal: false,
  sObjects: '<sObject name>',
  customObjects: false,
  returnPicklistDataAs: 'api',
  pollingInterval: '5 min',
  authToken: '<your auth token>',
  useBulkLoadV2: false,
  apiEndPoint: 'https://<your-instance>.my.salesforce.com/',
  MigrateSchema: false,
  useConnectionProfile: false,
  mode: 'Automated',
  connectionRetryPolicy: 'retryInterval=30, maxRetries=3',
  threadPoolSize: 5 )
OUTPUT TO SalesforceToClickhouse_OutputStream;

CREATE TARGET clickhousetgt2 USING Global.ClickHouseWriter (
  NullMarker: 'NULL',
  optimizedMerge: false,
  UploadPolicy: 'eventcount:10000,interval:60s',
  Tables: '<sObject>,<target_database>.<target_table>',
  CDDLAction: 'Process',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=10, totalTimeout=10m',
  ParallelThreads: 1,
  TableEngine: 'MergeTree',
  connectionProfileName: 'admin.ClickHouse_Connection_Profile',
  Mode: 'MERGE',
  DeleteStrategy: 'LightweightDelete' )
INPUT FROM SalesforceToClickhouse_OutputStream;

END FLOW SalesforceToClickhouse_SourceFlow;

END APPLICATION SalesforceToClickhouse;

MongoDB Reader emits a JSONNodeEvent with no type information, so a continuous query (CQ) is required to extract fields into a typed stream before ClickHouse Writer can map them to target columns. Because a typed stream supports INSERT only, use MergeTree with Mode set to APPENDONLY so that every incoming document is written as a new row.

CREATE OR REPLACE APPLICATION MongoToClickhouse USE EXCEPTIONSTORE TTL : '7d';

CREATE OR REPLACE SOURCE MongoToClickhouse_MongoDB USING MongoDBReader (
  collections: '<database>.<collection>',
  QuiesceOnILCompletion: true,
  Username: '<username>',
  Password: '<encrypted password>',
  Password_encrypted: 'true',
  ConnectionRetryPolicy: 'retryInterval=60, maxRetries=3',
  mode: 'InitialLoad',
  ConnectionURL: 'mongodb+srv://<cluster-host>/?appName=<app_name>' )
OUTPUT TO MongoToClickhouse_OutputStream;

CREATE OR REPLACE CQ JSON2StriimType
INSERT INTO jsonNodeEventStream
SELECT
  CASE WHEN IS_NULL(data.get('_id')) THEN null ELSE data.get('_id').toString() END,
  CASE WHEN IS_NULL(data.get('name')) THEN null ELSE data.get('name').toString() END,
  CASE WHEN IS_NULL(data.get('age')) THEN null ELSE data.get('age').toString() END
FROM MongoToClickhouse_OutputStream;

CREATE OR REPLACE TARGET clickhousetgt USING Global.ClickHouseWriter (
  NullMarker: 'NULL',
  DeleteStrategy: 'LightweightDelete',
  optimizedMerge: false,
  UploadPolicy: 'eventcount:10000,interval:30s',
  CDDLAction: 'Process',
  Tables: '<target_database>.emp',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=10, totalTimeout=10m',
  ParallelThreads: 1,
  connectionProfileName: 'admin.ClickHouse_Connection_Profile',
  Mode: 'APPENDONLY',
  TableEngine: 'MergeTree' )
INPUT FROM jsonNodeEventStream;

END APPLICATION MongoToClickhouse;

A Kafka source using a delimited (DSV) parser emits a WAEvent without a typeUUID, so define a Typed Stream (CREATE TYPE) and a CQ to parse and type each field before the event reaches ClickHouse Writer. Because Kafka is an append-only streaming source with no UPDATE or DELETE, MergeTree with Mode set to APPENDONLY is the recommended configuration, avoiding the overhead of mutation-based reconciliation.

CREATE OR REPLACE APPLICATION KafkaToClickHouse;

CREATE OR REPLACE SOURCE kafkasrc USING Global.KafkaReader (
  AutoMapPartition: true,
  startOffset: '0',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=5, totalTimeout=10m',
  connectionProfileName: 'admin.Kafka_Connection_Profile',
  Topic: '<topic name>' )
PARSE USING Global.DSVParser (
  columndelimiter: ',',
  trimquote: true,
  header: false )
OUTPUT TO clickhouseout;

CREATE TYPE EmpType (
  employee_id java.lang.Double,
  first_name  java.lang.String,
  last_name   java.lang.String,
  email       java.lang.String,
  hire_date   org.joda.time.DateTime,
  salary      java.lang.Double);

CREATE STREAM EmpStream OF EmpType;

CREATE OR REPLACE CQ EmpCQ
INSERT INTO EmpStream
SELECT TO_DOUBLE(data[6]),
       data[7],
       data[8],
       data[9],
       TO_DATE(data[10]),
       TO_DOUBLE(data[11])
FROM clickhouseout;

CREATE OR REPLACE TARGET clickhousetgt USING Global.ClickHouseWriter (
  DeleteStrategy: 'LightweightDelete',
  NullMarker: 'NULL',
  optimizedMerge: false,
  UploadPolicy: 'eventcount:10000,interval:30s',
  CDDLAction: 'Process',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=10, totalTimeout=10m',
  TableEngine: 'MergeTree',
  Mode: 'APPENDONLY',
  ParallelThreads: 1,
  Tables: '<target_database>.employees',
  connectionProfileName: 'admin.ClickHouse_Connection_Profile' )
INPUT FROM EmpStream;

END APPLICATION KafkaToClickHouse;

You can pair ClickHouse Writer with Striim's EMBEDDINGGENERATOR to keep a vector-search or RAG index continuously up to date as source rows change, instead of running a separate batch embedding pipeline. In this pattern, a CDC source (for example, MySQL Reader) feeds a continuous query that calls generateEmbeddings(...) against an embedding provider for each changed row, attaches the resulting vector to the event, and ClickHouse Writer appends the row, including its embedding, to a ClickHouse table. A downstream application then performs similarity search (for example, using ClickHouse's cosineDistance function) against the embedded column.

Configure an EMBEDDINGGENERATOR once, outside the application block, similar to a Connection Profile:

CREATE OR REPLACE EMBEDDINGGENERATOR OpenAIEmbedder USING OPENAI (
  apiKey:    '<YOUR_API_KEY>',
  modelName: 'text-embedding-3-small'
);

Then reference it from a continuous query that computes an embedding for the column you want to search on, and attaches it to the event as user data before it reaches ClickHouse Writer:

CREATE SOURCE MysqlSource USING Global.MysqlReader (
  CDDLCapture: true, connectionProfileName: 'admin.Connection_Profile',
  Tables: 'waction.drugs', SendBeforeImage: true )
OUTPUT TO sourceStream;

CREATE CQ GenerateEmbeddings
INSERT INTO embeddingStream
SELECT putUserData(e, 'embedding',
  java.util.Arrays.toString(
    generateEmbeddings("admin.OpenAIEmbedder",
      TO_STRING(GETDATA(e, "side_effects")))))
FROM sourceStream e;

CREATE TARGET ClickHouseWriter USING Global.ClickHouseWriter (
  Mode: 'APPENDONLY', TableEngine: 'MergeTree',
  Tables: 'waction.drugs,waction.drugs
    ColumnMap(embeddings=@USERDATA(embedding))',
  connectionProfileName: 'admin.ClickHouse_Connection_Profile' )
INPUT FROM embeddingStream;

On the ClickHouse side, add a String column to hold the embedding vector (Striim writes it as a bracketed, comma-separated string, for example [0.01234,-0.04567,...]) alongside the source table's normal columns:

CREATE TABLE waction.drugs (
  drug_name          String,
  medical_condition  String,
  side_effects       String,
  embeddings         String   -- Striim writes the vector as [f1,f2,...]
) ENGINE = MergeTree()
ORDER BY (medical_condition, drug_name);

Note: Run a separate initial-load application first to bulk-copy existing rows (initial load does not compute embeddings); to backfill embeddings for those historical rows, trigger a one-time source-side UPDATE after the initial load completes so the change flows through the embeddings CQ. Use ColumnMap in the Tables property to map the CQ's computed user-data field to the target embeddings column, and use Mode=APPENDONLY with TableEngine=MergeTree for the append-only embeddings write path shown here.

File Reader using a delimited (DSV) parser emits a WAEvent without a typeUUID, so define a Typed Stream (CREATE TYPE) and a CQ to parse and type each field — converting delimited text values to their target types with functions such as TO_DOUBLE and TO_DATE — before the event reaches ClickHouse Writer. Because a file source is typically a one-time or append-only load with no UPDATE or DELETE, MergeTree with Mode set to APPENDONLY is the recommended configuration.

CREATE OR REPLACE APPLICATION FileToClickHouse;

CREATE TYPE EmpType (
  employee_id java.lang.Double,
  first_name  java.lang.String,
  last_name   java.lang.String,
  email       java.lang.String,
  hire_date   org.joda.time.DateTime,
  salary      java.lang.Double);

CREATE OR REPLACE SOURCE FileSrc USING Global.FileReader (
  directory: '<path to source directory>',
  wildcard: 'employees_data.csv' )
PARSE USING Global.DSVParser (
  trimquote: true,
  header: true )
OUTPUT TO FileOut;

CREATE STREAM EmpStream OF EmpType;

CREATE CQ EmpCQ
INSERT INTO EmpStream
SELECT TO_DOUBLE(data[0]),
       data[1],
       data[2],
       data[3],
       TO_DATE(data[4]),
       TO_DOUBLE(data[5])
FROM FileOut;

CREATE OR REPLACE TARGET clickhousetgt USING Global.ClickHouseWriter (
  Mode: 'APPENDONLY',
  TableEngine: 'MergeTree',
  DeleteStrategy: 'LightweightDelete',
  Tables: '<target_database>.employees',
  connectionProfileName: 'admin.ClickHouse_Connection_Profile' )
INPUT FROM EmpStream;

END APPLICATION FileToClickHouse;