Prerequisites

1. Confluent Platform/Cloud with support for Confluent custom connector.
2. Registration of Confluent schemas in progress.
3. Access to a clickhouse data store.
4. Valid clickhouse credentials.
   
5. A valid Onibex license.
6. Onibex connector installed on Confluent Platform/Cloud.

• The URL to connect to the Clickhouse Cloud database, specifying the port and database.
  Example:
1.   jdbc:clickhouse://<cloud_host>:<clickhouse_port>/<database>?ssl=true
• Connection user: 
  
• JDBC user: Clickhouse Username.

• Private password:
  
  • JDBC password: Clickhouse Password.

6. Writes Settings:

• Insert mode: Registration insertion mode. Options: Insert.
• Batch size: Number of records grouped for SQL transactions.
• Enable Delete: Allows the deletion of records in the target database.  Options: True or False.
  

• The insert mode depend on the Primary key mode configuration (record_key or record_value) to accurately identify records.

• The Primary key mode, insert does not use keys to check for duplicates.

• The Enable deletions para meter depends on the Primary key mode set  of configuration a record_key.

• Key converter schema log URL: Confluent schema log URLs.
  Example:

• Value converter scheme record URL: URL of the Confluent Schemes Register for Values.
  Example:

1. ExtractTimestamp.type: Add a timestamp to the records.
   Example: org.apache.kafka.connect.transforms.InsertField$Value
2. transformes.ExtractTimestamp.timestamp.field: Field where the timestamp of the event will be inserted.
   Example: timestamp
3. transforms.InsertTimezone.type: Adds a time zone field to the records.
   Example: org.apache.kafka.connect.transforms.InsertField$Value
4. transforms.InsertTimezone.static.field: Static field for time zone.
   Example: timezone
5. transforms.InsertTimezone.static.value:Time zone field value.
   Example: America/Mexico_City

Limitations

1. Column auto-creation doesn’t support GENERATED ALWAYS AS expression. The default value for nullable columns is NULL.
2. Automatic table creation only supports the ENGINE = ReplacingMergeTree(_version)
   
3. The schema registry must have a value and key to perform insertions and deletions correctly.
   
4. The connector does not support String type topics, it is required that the value of each message has a structure (Struct), that is, it must have a schema with defined fields.
   
   

Data supported for the connector

Support for developments in the scheme

Name	Description	Values
auto.create	Allows automatic table creation if they do not exist	true / false
auto.evolve	Allows automatic evolution of tables if the outline changes	true / false

Converters

Name	Description	Values
key.converter	Key converter in Avro format	io.confluent.connect.avro.AvroConverter
value.converter	Value converter in Avro format	io.confluent.connect.avro.AvroConverter
.converter.schema.registry.url	Confluent Platform schema log URLs	http:/<schema_registry_ip>:<schema_registry_port>
.converter.schema.registry.url	Confluent Platform Log URL for Values	http:/<schema_registry_ip>:<schema_registry_port>

Transformations

Name	Description	Values
transformations	Transformations applied to data	ExtractTimestamp, InsertTimezone
ExtractTimestamp.type	Type of transformation to add the timestamp to records	org.apache.kafka.connect.transforms.InsertField$Value
transforms.ExtractTimestamp.timestamp.field	Field where the event timestamp will be inserted	timestamp
transforms.InsertTimezone.type	Type of transformation to add the time zone	org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertTimezone.static.field	Static field where the time zone will be inserted	timezone
transforms.InsertTimezone.static.value	Time zone field value	America/Mexico_City

---

Examples Clickhouse On-premise

Example of setup with schema avro and password logging (Confluent Platform)

{
 "value.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
 "key.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
"name": "<connector_name>",
"connector.class": "com.onibex.connect.clickhouse.jdbc.OnibexClickhouseSinkConnector",
 "tasks.max": "1",
 "key.converter": "io.confluent.connect.avro.AvroConverter",
 "value.converter": "io.confluent.connect.avro.AvroConverter",
 "transforms": "ExtractTimestamp",
 "topics": "<topic_name>",
 "transforms.ExtractTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value", "transforms.ExtractTimestamp.timestamp.field": "timestamp",
 "connection.url": "jdbc:clickhouse://<host/ip>:<port>/<database>",
 "connection.user": "<username_clickhouse>",
 "connection.password": "<password_clickhouse>",
 "insert.mode": "INSERT",
 "delete.enabled": "true",
 "table.name.format": "{$topic}",
 "pk.mode": "record_key",
 "auto.create": "true",
 "auto.evolve": "true",
 "onibex.license": "<valid_onibex_license>"
 }

Example of setup with schema Json and password logging (Confluent Platform)

{
 "value.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
 "key.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
"name": "<connector_name>",
"connector.class": "com.onibex.connect.clickhouse.jdbc.OnibexClickhouseSinkConnector",
 "tasks.max": "1",
 "key.converter": "io.confluent.connect.json.JsonSchemaConverter",
 "value.converter": "io.confluent.connect.json.JsonSchemaConverter",
 "transforms": "ExtractTimestamp",
 "topics": "<topic_name>",
 "transforms.ExtractTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",                                                                        
 "transforms.ExtractTimestamp.timestamp.field": "timestamp",
 "connection.url": "jdbc:clickhouse://<host/ip>:<port>/<database>",
 "connection.user": "<username_clickhouse>",
 "connection.password": "<password_clickhouse>",
 "insert.mode": "INSERT",
 "delete.enabled": "true",
 "table.name.format": "{$topic}",
 "pk.mode": "record_key",
 "auto.create": "true",
 "auto.evolve": "true",
 "onibex.license": "<valid_onibex_license>"
 }

Example of setup with schema avro  and password logging (Confluent Cloud)