VERSION

Features

  • Use InfluxDB sink connector to sync data from Pulsar

Tags

Pulsar IO

InfluxDB

Sink

Enterprise Support

StreamNative supported

Author

[ "ASF" ]

InfluxDB Sink

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

Name Type Required Default Description
influxdbUrl String true " " (empty string) The URL of the InfluxDB instance.
token String true " " (empty string) The authentication token used to authenticate to InfluxDB.
organization String true " " (empty string) The InfluxDB organization to write to.
bucket String true " " (empty string) The InfluxDB bucket to write to.
precision String false ns The timestamp precision for writing data to InfluxDB.

Below are the available options:
  • ns
  • us
  • ms
  • s
  • logLevel String false NONE The log level for InfluxDB request and response.

    Below are the available options:
  • NONE
  • BASIC
  • HEADERS
  • FULL
  • gzipEnable boolean false false Whether to enable gzip or not.
    batchTimeMs long false 1000L The InfluxDB operation time in milliseconds.
    batchSize int false 200 The batch size of writing to InfluxDB.

    InfluxDBv1

    Name Type Required Default Description
    influxdbUrl String true " " (empty string) The URL of the InfluxDB instance.
    username String false " " (empty string) The username used to authenticate to InfluxDB.
    password String false " " (empty string) The password used to authenticate to InfluxDB.
    database String true " " (empty string) The InfluxDB to which write messages.
    consistencyLevel String false ONE The consistency level for writing data to InfluxDB.

    Below are the available options:
  • ALL
  • ANY
  • ONE
  • QUORUM
  • logLevel String false NONE The log level for InfluxDB request and response.

    Below are the available options:
  • NONE
  • BASIC
  • HEADERS
  • FULL
  • retentionPolicy String false autogen The retention policy for InfluxDB.
    gzipEnable boolean false false Whether to enable gzip or not.
    batchTimeMs long false 1000L The InfluxDB operation time in milliseconds.
    batchSize int false 200 The 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

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

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

    InfluxDBv1

    • JSON

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

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