Kafka Destination Plugin

Example #

kind: destination
spec:
  name: "kafka"
  path: "cloudquery/kafka"
  registry: "cloudquery"
  version: "v5.4.17"
  write_mode: "append"
  spec:
    # required - list of brokers to connect to
    brokers: ["<broker-host>:<broker-port>"]
    # optional - if connecting via SASL/PLAIN, the username and password to use. If not set, no authentication will be used.
    sasl_username: "${KAFKA_SASL_USERNAME}"
    sasl_password: "${KAFKA_SASL_PASSWORD}"
    format: "json" # options: parquet, json, csv
    format_spec:
      # CSV specific parameters:
      # delimiter: ","
      # skip_header: false
      # Parquet specific parameters:
      # version: "v2Latest"
      # root_repetition: "repeated"
      # max_row_group_length: 134217728 # 128 * 1024 * 1024

    # Optional parameters
    # compression: "" # options: gzip
    # verbose: false
    # batch_size: 1000
    # topic_details:
      # num_partitions: 1
      # replication_factor: 1

Plugin Spec #

• brokers ([]string) (required)

  List of brokers to connect to.

  Example broker address:

  • "localhost:9092" default URL for a local Kafka broker
• format (string) (required)

  Format of the output file. Supported values are csv, json and parquet.
• format_spec (format_spec) (optional)

  Optional parameters to change the format of the file.
• compression (string) (optional) (default: empty)

  Compression algorithm to use. Supported values are empty or gzip. Not supported for parquet format.
• sasl_username (string) (optional) (default: empty)

  If connecting via SASL/PLAIN, the username to use.
• sasl_password (string) (optional) (default: empty)

  If connecting via SASL/PLAIN, the password to use.
• enforce_tls_verification (boolean) (optional) (default: false)

  If true, the plugin will verify the TLS certificate of the Kafka broker.
• verbose (boolean) (optional) (default: false)

  If true, the plugin will log all underlying Kafka client messages to the log.
• batch_size (integer) (optional) (default: 1000)

  Number of records to write before starting a new object.
• topic_details (topic_details) (optional)

  Optional parameters to set topic details.

format_spec #

CSV

• delimiter (string) (optional) (default: ,)

  Delimiter to use in the CSV file.
• skip_header (boolean) (optional) (default: false)

  If set to true, the CSV file will not contain a header row as the first row.

JSON

Parquet

• version (string) (optional) (default: v2Latest)

  Parquet format version to use. Supported values are v1.0, v2.4, v2.6 and v2Latest. v2Latest is an alias for the latest version available in the Parquet library which is currently v2.6.

  Useful when the reader consuming the Parquet files does not support the latest version.
• root_repetition (string) (optional) (default: repeated)

  Repetition option to use for the root node. Supported values are undefined, required, optional and repeated.

  Some Parquet readers require a specific root repetition option to be able to read the file. For example, importing Parquet files into Snowflake requires the root repetition to be undefined.
• max_row_group_length (integer) (optional) (default: 134217728 (= 128 * 1024 * 1024))

  The maximum number of rows in a single row group. Use a lower number to reduce memory usage when reading the Parquet files, and a higher number to increase the efficiency of reading the Parquet files.

topic_details #

• num_partitions (integer) (optional) (default: 1)

  Number of partitions for the newly created topic.
• replication_factor (integer) (optional) (default: 1)

  Replication factor for the topic.