Kafka and Clickhouse Integration

Powerful performance with an easy integration, powered by Telegraf, the open source data connector built by InfluxData.

info

This is not the recommended configuration for real-time query at scale. For query and compression optimization, high-speed ingest, and high availability, you may want to consider Kafka and InfluxDB.

5B+

Telegraf downloads

#1

Time series database
Source: DB Engines

1B+

Downloads of InfluxDB

2,800+

Contributors

Table of Contents

Powerful Performance, Limitless Scale

Collect, organize, and act on massive volumes of high-velocity data. Any data is more valuable when you think of it as time series data. with InfluxDB, the #1 time series platform built to scale with Telegraf.

See Ways to Get Started

Input and output integration overview

The Kafka plugin reads from Kafka and creates metrics using one of the supported input data formats.

Telegraf’s SQL output plugin sends collected metrics to an SQL database using a straightforward table schema and dynamic column generation. When configured for ClickHouse, it adjusts DSN formatting and type conversion settings to ensure seamless data integration.

Integration details

Kafka

The Kafka plugin allows you to read messages from Kafka topics and create metrics. It supports various features, including SASL authentication, message headers as tags, and different message consumption strategies.

Clickhouse

The SQL output plugin is designed to store Telegraf metrics in an SQL database using a simple, hard-coded schema. Each metric type gets its own table, and columns are generated for every tag and field, with an optional timestamp column. For ClickHouse, the plugin leverages a specialized DSN format as defined by clickhouse-go v1.5.4 and customizes metric type conversion to align with ClickHouse data types. This ensures that integers, texts, timestamps, booleans, and real numbers are mapped to ClickHouse’s native types such as Int64, String, DateTime, UInt8, and Float64 respectively.

Configuration

Kafka


[[inputs.kafka_consumer]]
              ## Kafka brokers.
              brokers = ["localhost:9092"]

              ## Set the minimal supported Kafka version. Should be a string contains
              ## 4 digits in case if it is 0 version and 3 digits for versions starting
              ## from 1.0.0 separated by dot. This setting enables the use of new
              ## Kafka features and APIs.  Must be 0.10.2.0(used as default) or greater.
              ## Please, check the list of supported versions at
              ## https://pkg.go.dev/github.com/Shopify/sarama#SupportedVersions
              ##   ex: kafka_version = "2.6.0"
              ##   ex: kafka_version = "0.10.2.0"
              # kafka_version = "0.10.2.0"

              ## Topics to consume.
              topics = ["telegraf"]

              ## Topic regular expressions to consume.  Matches will be added to topics.
              ## Example: topic_regexps = [ "*test", "metric[0-9A-z]*" ]
              # topic_regexps = [ ]

              ## When set this tag will be added to all metrics with the topic as the value.
              # topic_tag = ""

              ## The list of Kafka message headers that should be pass as metric tags
              ## works only for Kafka version 0.11+, on lower versions the message headers
              ## are not available
              # msg_headers_as_tags = []

              ## The name of kafka message header which value should override the metric name.
              ## In case when the same header specified in current option and in msg_headers_as_tags
              ## option, it will be excluded from the msg_headers_as_tags list.
              # msg_header_as_metric_name = ""

              ## Set metric(s) timestamp using the given source.
              ## Available options are:
              ##   metric -- do not modify the metric timestamp
              ##   inner  -- use the inner message timestamp (Kafka v0.10+)
              ##   outer  -- use the outer (compressed) block timestamp (Kafka v0.10+)
              # timestamp_source = "metric"

              ## Optional Client id
              # client_id = "Telegraf"

              ## Optional TLS Config
              # enable_tls = false
              # tls_ca = "/etc/telegraf/ca.pem"
              # tls_cert = "/etc/telegraf/cert.pem"
              # tls_key = "/etc/telegraf/key.pem"
              ## Use TLS but skip chain & host verification
              # insecure_skip_verify = false

              ## Period between keep alive probes.
              ## Defaults to the OS configuration if not specified or zero.
              # keep_alive_period = "15s"

              ## SASL authentication credentials.  These settings should typically be used
              ## with TLS encryption enabled
              # sasl_username = "kafka"
              # sasl_password = "secret"

              ## Optional SASL:
              ## one of: OAUTHBEARER, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI
              ## (defaults to PLAIN)
              # sasl_mechanism = ""

              ## used if sasl_mechanism is GSSAPI
              # sasl_gssapi_service_name = ""
              # ## One of: KRB5_USER_AUTH and KRB5_KEYTAB_AUTH
              # sasl_gssapi_auth_type = "KRB5_USER_AUTH"
              # sasl_gssapi_kerberos_config_path = "/"
              # sasl_gssapi_realm = "realm"
              # sasl_gssapi_key_tab_path = ""
              # sasl_gssapi_disable_pafxfast = false

              ## used if sasl_mechanism is OAUTHBEARER
              # sasl_access_token = ""

              ## SASL protocol version.  When connecting to Azure EventHub set to 0.
              # sasl_version = 1

              # Disable Kafka metadata full fetch
              # metadata_full = false

              ## Name of the consumer group.
              # consumer_group = "telegraf_metrics_consumers"

              ## Compression codec represents the various compression codecs recognized by
              ## Kafka in messages.
              ##  0 : None
              ##  1 : Gzip
              ##  2 : Snappy
              ##  3 : LZ4
              ##  4 : ZSTD
              # compression_codec = 0
              ## Initial offset position; one of "oldest" or "newest".
              # offset = "oldest"

              ## Consumer group partition assignment strategy; one of "range", "roundrobin" or "sticky".
              # balance_strategy = "range"

              ## Maximum number of retries for metadata operations including
              ## connecting. Sets Sarama library's Metadata.Retry.Max config value. If 0 or
              ## unset, use the Sarama default of 3,
              # metadata_retry_max = 0

              ## Type of retry backoff. Valid options: "constant", "exponential"
              # metadata_retry_type = "constant"

              ## Amount of time to wait before retrying. When metadata_retry_type is
              ## "constant", each retry is delayed this amount. When "exponential", the
              ## first retry is delayed this amount, and subsequent delays are doubled. If 0
              ## or unset, use the Sarama default of 250 ms
              # metadata_retry_backoff = 0

              ## Maximum amount of time to wait before retrying when metadata_retry_type is
              ## "exponential". Ignored for other retry types. If 0, there is no backoff
              ## limit.
              # metadata_retry_max_duration = 0

              ## When set to true, this turns each bootstrap broker address into a set of
              ## IPs, then does a reverse lookup on each one to get its canonical hostname.
              ## This list of hostnames then replaces the original address list.
              ## resolve_canonical_bootstrap_servers_only = false

              ## Strategy for making connection to kafka brokers. Valid options: "startup",
              ## "defer". If set to "defer" the plugin is allowed to start before making a
              ## connection. This is useful if the broker may be down when telegraf is
              ## started, but if there are any typos in the broker setting, they will cause
              ## connection failures without warning at startup
              # connection_strategy = "startup"

              ## Maximum length of a message to consume, in bytes (default 0/unlimited);
              ## larger messages are dropped
              max_message_len = 1000000

              ## Max undelivered messages
              ## This plugin uses tracking metrics, which ensure messages are read to
              ## outputs before acknowledging them to the original broker to ensure data
              ## is not lost. This option sets the maximum messages to read from the
              ## broker that have not been written by an output.
              ##
              ## This value needs to be picked with awareness of the agent's
              ## metric_batch_size value as well. Setting max undelivered messages too high
              ## can result in a constant stream of data batches to the output. While
              ## setting it too low may never flush the broker's messages.
              # max_undelivered_messages = 1000

              ## Maximum amount of time the consumer should take to process messages. If
              ## the debug log prints messages from sarama about 'abandoning subscription
              ## to [topic] because consuming was taking too long', increase this value to
              ## longer than the time taken by the output plugin(s).
              ##
              ## Note that the effective timeout could be between 'max_processing_time' and
              ## '2 * max_processing_time'.
              # max_processing_time = "100ms"

              ## The default number of message bytes to fetch from the broker in each
              ## request (default 1MB). This should be larger than the majority of
              ## your messages, or else the consumer will spend a lot of time
              ## negotiating sizes and not actually consuming. Similar to the JVM's
              ## `fetch.message.max.bytes`.
              # consumer_fetch_default = "1MB"

              ## Data format to consume.
              ## Each data format has its own unique set of configuration options, read
              ## more about them here:
              ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
              data_format = "influx"

Clickhouse

[[outputs.sql]]
  ## Database driver
  ## Valid options include mssql, mysql, pgx, sqlite, snowflake, clickhouse
  driver = "clickhouse"

  ## Data source name
  ## For ClickHouse, the DSN follows the clickhouse-go v1.5.4 format.
  ## Example DSN: "tcp://localhost:9000?debug=true"
  data_source_name = "tcp://localhost:9000?debug=true"

  ## Timestamp column name
  timestamp_column = "timestamp"

  ## Table creation template
  ## Available template variables:
  ##  {TABLE}        - table name as a quoted identifier
  ##  {TABLELITERAL} - table name as a quoted string literal
  ##  {COLUMNS}      - column definitions (list of quoted identifiers and types)
  table_template = "CREATE TABLE {TABLE} ({COLUMNS})"

  ## Table existence check template
  ## Available template variables:
  ##  {TABLE} - table name as a quoted identifier
  table_exists_template = "SELECT 1 FROM {TABLE} LIMIT 1"

  ## Initialization SQL (optional)
  init_sql = ""

  ## Maximum amount of time a connection may be idle. "0s" means connections are never closed due to idle time.
  connection_max_idle_time = "0s"

  ## Maximum amount of time a connection may be reused. "0s" means connections are never closed due to age.
  connection_max_lifetime = "0s"

  ## Maximum number of connections in the idle connection pool. 0 means unlimited.
  connection_max_idle = 2

  ## Maximum number of open connections to the database. 0 means unlimited.
  connection_max_open = 0

  ## Metric type to SQL type conversion for ClickHouse.
  ## The conversion maps Telegraf metric types to ClickHouse native data types.
  [outputs.sql.convert]
    conversion_style = "literal"
    integer          = "Int64"
    text             = "String"
    timestamp        = "DateTime"
    defaultvalue     = "String"
    unsigned         = "UInt64"
    bool             = "UInt8"
    real             = "Float64"

Input and output integration examples

Kafka

  1. Real-Time Data Processing: Use the Kafka Consumer Input Plugin to read data from Kafka topics in real-time, allowing for immediate metrics generation and processing.
  2. SASL Authentication: Configure the plugin with SASL authentication to securely connect to Kafka brokers, ensuring that only authorized users can access the data.
  3. Multiple Topic Consumption: Set up the plugin to consume from multiple Kafka topics by specifying them in the configuration. This allows you to gather metrics from various data sources simultaneously.
  4. Message Transformation: Leverage the plugin’s ability to parse and transform messages into metrics based on the specified data_format, enabling tailored data handling for your specific use case.

Clickhouse

  1. Basic ClickHouse Integration: Configure the plugin by setting the driver to ‘clickhouse’ and providing the appropriate DSN format as required by clickhouse-go v1.5.4. This ensures that Telegraf can connect and write metrics to your ClickHouse database.

  2. Customized Table Schemas: Leverage the table creation and existence check templates to tailor the database schema. This allows you to predefine column types and even disable automatic table creation if you prefer manual schema management.

  3. Advanced Type Conversion: Utilize the ClickHouse-specific conversion settings to map Telegraf metric types directly to ClickHouse data types (e.g., mapping integers to Int64 and timestamps to DateTime). This ensures data is stored with the correct precision and format.

  4. Initialization and Connection Tuning: Use the init_sql setting to run custom SQL commands upon connection, and adjust connection pool settings (like connection_max_idle_time and connection_max_open) to optimize performance for high-throughput environments.

Feedback

Thank you for being part of our community! If you have any general feedback or found any bugs on these pages, we welcome and encourage your input. Please submit your feedback in the InfluxDB community Slack.

Powerful Performance, Limitless Scale

Collect, organize, and act on massive volumes of high-velocity data. Any data is more valuable when you think of it as time series data. with InfluxDB, the #1 time series platform built to scale with Telegraf.

See Ways to Get Started

Related Integrations

HTTP and InfluxDB Integration

The HTTP plugin collects metrics from one or more HTTP(S) endpoints. It supports various authentication methods and configuration options for data formats.

View Integration

Kafka and InfluxDB Integration

This plugin reads messages from Kafka and allows the creation of metrics based on those messages. It supports various configurations including different Kafka settings and message processing options.

View Integration

Kinesis and InfluxDB Integration

The Kinesis plugin allows for reading metrics from AWS Kinesis streams. It supports multiple input data formats and offers checkpointing features with DynamoDB for reliable message processing.

View Integration