Skip to main content
Version: 3.3.x

InfluxDB sink connector

note

You can download all the Pulsar connectors on download page.

The InfluxDB sink connector pulls messages from Pulsar topics and persists the messages to InfluxDB.

The InfluxDB sink provides different configurations for InfluxDBv1 and v2 respectively.

Configuration

The configuration of the InfluxDB sink connector has the following properties.

Property

InfluxDBv2

NameTypeRequiredDefaultDescription
influxdbUrlStringtrue" " (empty string)The URL of the InfluxDB instance.
tokenStringtrue" " (empty string)The authentication token used to authenticate to InfluxDB.
organizationStringtrue" " (empty string)The InfluxDB organization to write to.
bucketStringtrue" " (empty string)The InfluxDB bucket to write to.
precisionStringfalsensThe timestamp precision for writing data to InfluxDB.

Below are the available options:
  • ns
  • us
  • ms
  • s
  • logLevelStringfalseNONEThe log level for InfluxDB request and response.

    Below are the available options:
  • NONE
  • BASIC
  • HEADERS
  • FULL
  • gzipEnablebooleanfalsefalseWhether to enable gzip or not.
    batchTimeMslongfalse1000LThe InfluxDB operation time in milliseconds.
    batchSizeintfalse200The batch size of writing to InfluxDB.

    InfluxDBv1

    NameTypeRequiredDefaultDescription
    influxdbUrlStringtrue" " (empty string)The URL of the InfluxDB instance.
    usernameStringfalse" " (empty string)The username used to authenticate to InfluxDB.
    passwordStringfalse" " (empty string)The password used to authenticate to InfluxDB.
    databaseStringtrue" " (empty string)The InfluxDB to which write messages.
    consistencyLevelStringfalseONEThe consistency level for writing data to InfluxDB.

    Below are the available options:
  • ALL
  • ANY
  • ONE
  • QUORUM
  • logLevelStringfalseNONEThe log level for InfluxDB request and response.

    Below are the available options:
  • NONE
  • BASIC
  • HEADERS
  • FULL
  • retentionPolicyStringfalseautogenThe retention policy for InfluxDB.
    gzipEnablebooleanfalsefalseWhether to enable gzip or not.
    batchTimeMslongfalse1000LThe InfluxDB operation time in milliseconds.
    batchSizeintfalse200The batch size of writing to InfluxDB.

    Example

    Before using the InfluxDB sink connector, you need to create a configuration file through one of the following methods.

    InfluxDBv2

    • JSON

      {
      "configs": {
      "influxdbUrl": "http://localhost:9999",
      "organization": "example-org",
      "bucket": "example-bucket",
      "token": "xxxx",
      "precision": "ns",
      "logLevel": "NONE",
      "gzipEnable": false,
      "batchTimeMs": 1000,
      "batchSize": 100
      }
      }
    • YAML

      configs:
      influxdbUrl: "http://localhost:9999"
      organization: "example-org"
      bucket: "example-bucket"
      token: "xxxx"
      precision: "ns"
      logLevel: "NONE"
      gzipEnable: false
      batchTimeMs: 1000
      batchSize: 100

    InfluxDBv1

    • JSON

      {
      "configs": {
      "influxdbUrl": "http://localhost:8086",
      "database": "test_db",
      "consistencyLevel": "ONE",
      "logLevel": "NONE",
      "retentionPolicy": "autogen",
      "gzipEnable": false,
      "batchTimeMs": 1000,
      "batchSize": 100
      }
      }
    • YAML

      configs:
      influxdbUrl: "http://localhost:8086"
      database: "test_db"
      consistencyLevel: "ONE"
      logLevel: "NONE"
      retentionPolicy: "autogen"
      gzipEnable: false
      batchTimeMs: 1000
      batchSize: 100