InfluxDB

InfluxDB sink connector

Description

Write SeaTunnel rows to InfluxDB 1.x. The sink converts one row into one InfluxDB point, using measurement, key_time, and key_tags to decide the target measurement, timestamp, tags, and fields.

Key features

• exactly-once
• support multiple table write

Data Type Mapping

SeaTunnel Data Type	InfluxDB Usage
BOOLEAN	Written as an InfluxDB field.
SMALLINT	Written as an InfluxDB field.
INT	Written as an InfluxDB field.
BIGINT	Written as an InfluxDB field, or as the timestamp when configured by key_time.
FLOAT	Written as an InfluxDB field.
DOUBLE	Written as an InfluxDB field.
STRING	Written as an InfluxDB field, a tag value, or a timestamp string when configured by key_time.
TIMESTAMP	Can be used by key_time; converted with UTC zone to epoch milliseconds.

Other SeaTunnel types are not supported by the current InfluxDB sink serializer.

Options

name	type	required	default value	description
url	string	yes	-	InfluxDB server URL, for example http://influxdb-host:8086.
database	string	yes	-	InfluxDB database name.
measurement	string	no	input table full name	InfluxDB measurement name. If it is not configured, the sink uses the input table name.
username	string	no	-	InfluxDB username. It must be configured together with password.
password	string	no	-	InfluxDB password. It must be configured together with username.
key_time	string	no	processing time	Field name used as the InfluxDB point timestamp. If omitted, processing time is used.
key_tags	array	no	-	Field names written as InfluxDB tags. Other fields are written as point fields.
batch_size	int	no	1024	Number of points buffered before flushing to InfluxDB.
max_retries	int	no	-	Maximum retry count when flushing points fails.
write_timeout	int	no	5	Write timeout used by the InfluxDB client.
retry_backoff_multiplier_ms	int	no	-	Backoff multiplier used between retry attempts, in milliseconds.
max_retry_backoff_ms	int	no	-	Maximum backoff between retry attempts, in milliseconds.
rp	string	no	-	Retention policy used when writing points.
epoch	string	no	n	Time precision used by the client. Uppercase values H, M, S, MS, U, and NS are recognized for write precision.
connect_timeout_ms	long	no	15000	Timeout for connecting to InfluxDB, in milliseconds.
query_timeout_sec	int	no	3	Read timeout used by the InfluxDB client, in seconds.
multi_table_sink_replica	int	no	-	Replica count for multi-table sink writers.
common-options	config	no	-	Sink plugin common options.

url

the url to connect to influxDB e.g.

http://influxdb-host:8086

database [string]

The name of influxDB database

measurement [string]

The name of influxDB measurement. This option is optional. If it is omitted, the sink uses the input table full name as the measurement name, which is useful for multi-table writes. For multi-table input, make sure the generated table names are valid InfluxDB measurement names.

username [string]

influxDB user username

password [string]

influxDB user password

key_time [string]

Specify field-name of the influxDB measurement timestamp in SeaTunnelRow. If not specified, use processing-time as timestamp

key_tags [array]

Specify field-name of the influxDB measurement tags in SeaTunnelRow. If not specified, include all fields with influxDB measurement field

batch_size [int]

For batch writing, when the number of buffers reaches the number of batch_size or the time reaches checkpoint.interval, the data will be flushed into the influxDB

max_retries [int]

The number of retries to flush failed

If this option is not configured, the sink tries the write once and fails immediately if that write fails.

retry_backoff_multiplier_ms [int]

Using as a multiplier for generating the next delay for backoff

max_retry_backoff_ms [int]

The amount of time to wait before attempting to retry a request to influxDB

When retry options are used, configure both retry_backoff_multiplier_ms and max_retry_backoff_ms; otherwise the retry sleep interval remains 0.

write_timeout [int]

The write timeout used by the InfluxDB client.

rp [string]

Retention policy used when writing points.

epoch [string]

Time precision used by the InfluxDB client. For sink write precision, uppercase values H, M, S, MS, U, and NS are recognized by the current connector. The default n is treated as nanosecond precision.

query_timeout_sec [int]

The read timeout used by the InfluxDB client, in seconds.

connect_timeout_ms [long]

the timeout for connecting to InfluxDB, in milliseconds

common options

Sink plugin common parameters, please refer to Sink Common Options for details

Examples

Write One Measurement

sink {
    InfluxDB {
        url = "http://influxdb-host:8086"
        database = "test"
        measurement = "sink"
        key_time = "time"
        key_tags = ["label"]
        batch_size = 1
    }
}

Write Without Explicit Measurement

When measurement is omitted, the sink uses the input table full name as the measurement name.

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      table = "influxdb_sink"
      fields {
        label = STRING
        c_string = STRING
        c_double = DOUBLE
        c_bigint = BIGINT
        c_float = FLOAT
        c_int = INT
        c_smallint = SMALLINT
        c_boolean = BOOLEAN
        time = BIGINT
      }
    }
    rows = [
      {
        kind = INSERT
        fields = ["label_1", "sink_1", 4.3, 200, 2.5, 2, 5, true, 1627529632356]
      }
    ]
  }
}

sink {
  InfluxDB {
    url = "http://influxdb-host:8086"
    database = "test"
    key_time = "time"
    key_tags = ["label"]
    batch_size = 1
  }
}

Multiple Table

When measurement is omitted, each upstream table is written to a measurement named after that table. This is the usual setting for multi-table input.

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  Mysql-CDC {
    url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
    username = "root"
    password = "******"
    
    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]
  }
}

transform {
}

sink {
  InfluxDB {
    url = "http://influxdb-host:8086"
    database = "test"
    key_time = "time"
    batch_size = 1
  }
}

Changelog

Change Log

Change	Commit	Version
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Improve] influxdb options (#8966)	https://github.com/apache/seatunnel/commit/9f498b8133	2.3.10
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Improve] Improve some connectors prepare check error message (#7465)	https://github.com/apache/seatunnel/commit/6930a25edd	2.3.8
[Improve][Connector] Add multi-table sink option check (#7360)	https://github.com/apache/seatunnel/commit/2489f6446b	2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)	https://github.com/apache/seatunnel/commit/c4ca74122c	2.3.6
Support multi-table sink feature for influxdb (#6278)	https://github.com/apache/seatunnel/commit/56f13e920d	2.3.5
[Improve][Zeta] Add classloader cache mode to fix metaspace leak (#6355)	https://github.com/apache/seatunnel/commit/9c3c2f183d	2.3.5
[Test][E2E] Add thread leak check for connector (#5773)	https://github.com/apache/seatunnel/commit/1f2f3fc5f0	2.3.4
[BugFix][InfluxDBSource] Resolve invalid SQL in initColumnsIndex method caused by direct QUERY_LIMIT appendage with 'tz' function. (#4829)	https://github.com/apache/seatunnel/commit/deed9c62c3	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Improve][Connector-V2] Remove scheduler in InfluxDB sink (#5271)	https://github.com/apache/seatunnel/commit/f459f500cb	2.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)	https://github.com/apache/seatunnel/commit/51c0d709ba	2.3.4
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Improve][SourceConnector] Unifie InfluxDB source fields to schema (#3897)	https://github.com/apache/seatunnel/commit/85a984a64f	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)	https://github.com/apache/seatunnel/commit/b9164b8ba1	2.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Improve][Connector-V2][Influxdb] Unified exception for influxdb source & sink connector (#3558)	https://github.com/apache/seatunnel/commit/4686f35d68	2.3.0
[Feature][Connector][influx] Expose configurable options in influx db (#3392)	https://github.com/apache/seatunnel/commit/b247ff0aef	2.3.0
[Feature][Connector-V2] influxdb sink connector (#3174)	https://github.com/apache/seatunnel/commit/630e884791	2.3.0
[Feature][Connector-V2] Add influxDB connector source (#2697)	https://github.com/apache/seatunnel/commit/1d70ea3084	2.3.0-beta