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

NameTypeRequiredSensitiveDefaultDescription
influxdbUrlStringtruefalse" " (empty string)The URL of the InfluxDB instance.
tokenStringtruetrue" " (empty string)The authentication token used to authenticate to InfluxDB.
organizationStringtruefalse" " (empty string)The InfluxDB organization to write to.
bucketStringtruefalse" " (empty string)The InfluxDB bucket to write to.
precisionStringfalsefalsensThe 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
logLevelStringfalsefalseNONEThe 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
gzipEnablebooleanfalsefalsefalseWhether to enable gzip or not.
batchTimeMslongfalsefalse1000LThe InfluxDB operation time in milliseconds.
batchSizeintfalsefalse200The batch size of writing to InfluxDB.

InfluxDBv1

NameTypeRequiredSensitiveDefaultDescription
influxdbUrlStringtruefalse" " (empty string)The URL of the InfluxDB instance.
usernameStringfalsetrue" " (empty string)The username used to authenticate to InfluxDB.
passwordStringfalsetrue" " (empty string)The password used to authenticate to InfluxDB.
databaseStringtruefalse" " (empty string)The InfluxDB to which write messages.
consistencyLevelStringfalsefalseONEThe 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
logLevelStringfalsefalseNONEThe 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
retentionPolicyStringfalsefalseautogenThe retention policy for InfluxDB.
gzipEnablebooleanfalsefalsefalseWhether to enable gzip or not.
batchTimeMslongfalsefalse1000LThe InfluxDB operation time in milliseconds.
batchSizeintfalsefalse200The 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
    }