InfluxDB

InfluxDB sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

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
• cdc
• support multiple table write
• timer flush

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.

Sink 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. See Sink Common Options.

url [string]

The URL to connect to InfluxDB, for example http://influxdb-host:8086.

database [string]

The name of the InfluxDB database that points are written to.

measurement [string]

The InfluxDB measurement name. When omitted, the sink uses the input table full name as the measurement — which is the common setting for multi-table writes. Make sure the generated table names are valid InfluxDB measurement names.

username [string]

The InfluxDB user name. Configure it together with password when the influxdb requires authentication.

password [string]

The InfluxDB user password. Configure it together with username when authentication is required.

key_time [string]

The field name in the upstream row that supplies the InfluxDB point timestamp. When omitted, the sink uses the current processing time. Values configured here are accepted as numeric timestamps or as an ISO-8601 timestamp string.

key_tags [array]

The field names in the upstream row that should be written as InfluxDB tags. All other fields are written as point fields. When omitted, every field is written as a point field.

batch_size [int]

Number of points buffered before flushing to InfluxDB. The default is 1024. The buffer is also flushed at each checkpoint and when the writer closes.

max_retries [int]

Maximum retry count when the flush fails. When this option is not set, the sink writes once and fails immediately if that write fails.

retry_backoff_multiplier_ms [int]

Backoff multiplier used between retry attempts, in milliseconds. Configure it together with max_retry_backoff_ms so the retry sleep interval is non-zero.

max_retry_backoff_ms [int]

Maximum backoff between retry attempts, in milliseconds. Configure it together with retry_backoff_multiplier_ms so the retry sleep interval is non-zero.

write_timeout [int]

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

rp [string]

Retention policy used when writing points.

epoch [string]

Time precision used by the InfluxDB client. Uppercase values H, M, S, MS, U, and NS are recognized for write precision. The default value 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. The default is 15000.

common options

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

Task Example

Write One Measurement

A simple sink that writes rows to a single named 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
  }
}

Multi-Table Write

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