sink
InfluxDB Sink
The InfluxDB sink connector pulls messages from Pulsar topics and persists the messages to InfluxDB.
Authored by
ASF
Support type
StreamNative
License
Apache License 2.0

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. <br><br>Below are the available options:<li>ns<br><li>us<br><li>ms<br><li>s
logLevelStringfalseNONEThe log level for InfluxDB request and response. <br><br>Below are the available options:<li>NONE<br><li>BASIC<br><li>HEADERS<br><li>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. <br><br>Below are the available options:<li>ALL<br><li> ANY<br><li>ONE<br><li>QUORUM
logLevelStringfalseNONEThe log level for InfluxDB request and response. <br><br>Below are the available options:<li>NONE<br><li>BASIC<br><li>HEADERS<br><li>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

    {
        "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
    }