InfluxDB

InfluxDB source connector

Description

Read data from InfluxDB 1.x by using an InfluxQL query. The connector supports a normal single query and an optional parallel scan mode that splits one query by an integer column range.

Key features

• batch
• stream
• exactly-once
• column projection
• parallelism
• support user-defined split

Data Type Mapping

SeaTunnel Data Type	Notes
BOOLEAN	Parsed from the returned InfluxDB value.
SMALLINT	Parsed from the returned InfluxDB value.
INT	Parsed from the returned InfluxDB value.
BIGINT	Parsed from the returned InfluxDB value.
FLOAT	InfluxDB returns numbers as double values; the connector converts them to FLOAT.
DOUBLE	Uses the returned numeric value.
STRING	Uses the returned value as a string.

Other SeaTunnel types are not supported by the current InfluxDB source converter.

Options

name	type	required	default value	description
url	string	yes	-	InfluxDB server URL, for example http://influxdb-host:8086.
sql	string	yes	-	InfluxQL query used to read data.
schema	config	yes	-	Output schema returned by the source.
database	string	yes	-	InfluxDB database 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.
lower_bound	int	no	-	Lower bound of split_column when parallel scan is enabled.
upper_bound	int	no	-	Upper bound of split_column when parallel scan is enabled.
partition_num	int	no	0	Number of query splits. 0 means the source runs the original sql as one split.
split_column	string	no	-	Integer column used to split the query when parallel scan is enabled.
where	string	no	-	Reserved source option. The current split logic reads the lowercase where keyword from sql directly.
epoch	string	no	n	Time precision returned by InfluxDB. For example: H, m, s, MS, u, n.
connect_timeout_ms	long	no	15000	Timeout for connecting to InfluxDB, in milliseconds.
query_timeout_sec	int	no	3	Timeout for querying InfluxDB, in seconds.
common-options	config	no	-	Source plugin common options.

url

the url to connect to influxDB e.g.

http://influxdb-host:8086

sql [string]

The query sql used to search data

select name,age from test

schema [config]

fields [Config]

The schema information of upstream data. For more details, please refer to Schema Feature. e.g.

schema {
    fields {
        name = string
        age = int
    }
  }

database [string]

The influxDB database

username [string]

the username of the influxDB when you select

password [string]

the password of the influxDB when you select

split_column [string]

The column used to split one query into multiple range queries.

Tips:

• influxDB tags is not supported as a segmented primary key because the type of tags can only be a string
• influxDB time is not supported as a segmented primary key because the time field cannot participate in mathematical calculation
• Currently, split_column only supports integer data segmentation, and does not support float, string, date and other types.
• split_column, lower_bound, upper_bound, and partition_num must be configured together.
• If the split query contains a filter, use lowercase where in sql, for example select * from test where age > 0. The current split parser is case-sensitive.
• where is an option in the validation rule, but the current split logic reads the filter from sql. Put the filter in sql instead of configuring a separate where value.

upper_bound [int]

upper bound of the split_columncolumn

lower_bound [int]

lower bound of the split_column column

     split the $split_column range into $partition_num parts
     if partition_num is 1, use the whole `split_column` range
     if partition_num < (upper_bound - lower_bound), use (upper_bound - lower_bound) partitions
     
     eg: lower_bound = 1, upper_bound = 10, partition_num = 2
     sql = "select * from test where age > 0 and age < 10"
     
     split result

     split 1: select * from test where ($split_column >= 1 and $split_column < 6)  and (  age > 0 and age < 10 )
     
     split 2: select * from test where ($split_column >= 6 and $split_column < 11) and (  age > 0 and age < 10 )

partition_num [int]

the partition_num of the InfluxDB when you select

Tips: Ensure that upper_bound minus lower_bound is divided bypartition_num, otherwise the query results will overlap

epoch [string]

returned time precision

• Optional values: H, m, s, MS, u, n
• default value: n

query_timeout_sec [int]

the query_timeout of the InfluxDB when you select, in seconds

connect_timeout_ms [long]

the timeout for connecting to InfluxDB, in milliseconds

common options

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

Examples

Read With Parallel Range Splits

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

source {

    InfluxDB {
        url = "http://influxdb-host:8086"
        sql = "select label, c_string, c_double, c_bigint, c_float, c_int, c_smallint, c_boolean from source"
        database = "test"
        upper_bound = 99
        lower_bound = 0
        partition_num = 4
        split_column = "c_int"
        schema {
            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
            }
        }
    }

}

sink {
    Console {}
}

Read Without Parallel Range Splits

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

source {

    InfluxDB {
        url = "http://influxdb-host:8086"
        sql = "select label, c_string, c_double, c_bigint, c_float, c_int, c_smallint, c_boolean from source"
        database = "test"
        schema {
            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
            }
        }
    }

}

sink {
    Console {}
}

Read With InfluxQL Time Zone

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

source {
    InfluxDB {
        url = "http://influxdb-host:8086"
        sql = "select label, c_string, c_double, c_bigint, c_float, c_int, c_smallint, c_boolean from source tz('Asia/Shanghai')"
        database = "test"
        schema {
            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
            }
        }
    }
}

sink {
    Console {}
}

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