跳到主要内容
版本：Next

InfluxDB

InfluxDB 源连接器

引擎支持​

Spark
Flink
SeaTunnel Zeta

描述​

通过 InfluxQL 查询从 InfluxDB 1.x 读取数据。连接器支持普通单查询，也支持按一个整数列范围切分查询， 让多个并行任务分别读取不同范围的数据。

关键特性​

数据类型映射​

SeaTunnel 数据类型说明
BOOLEAN从 InfluxDB 返回值解析。
SMALLINT从 InfluxDB 返回值解析。
INT从 InfluxDB 返回值解析。
BIGINT从 InfluxDB 返回值解析。
FLOATInfluxDB 会把数字按 double 返回，连接器再转换成 FLOAT。
DOUBLE使用返回的数字值。
STRING使用返回值作为字符串。

当前 InfluxDB source 转换器不支持其他 SeaTunnel 类型。

Source 选项​

参数名类型必须默认值描述
urlstring是-InfluxDB 连接 URL，例如 http://influxdb-host:8086。
sqlstring是-用于读取数据的 InfluxQL 查询。
schemaconfig是-上游数据的 schema 信息。更多详情请参考 Schema 特性。
databasestring是-InfluxDB 数据库名称。
usernamestring否-InfluxDB 用户名。必须和 password 一起配置。
passwordstring否-InfluxDB 密码。必须和 username 一起配置。
lower_boundint否-启用并行范围读取时，split_column 的下界。
upper_boundint否-启用并行范围读取时，split_column 的上界。
partition_numint否0查询切分数量。0 表示不切分，直接执行原始 sql。
split_columnstring否-用于并行切分的整数列。
wherestring否-预留的 source 配置项。当前切分逻辑直接从 sql 中读取小写 where 关键字。
epochstring否n返回的时间精度，例如：H、m、s、MS、u、n。
connect_timeout_mslong否15000连接 InfluxDB 的超时时间（毫秒）。
query_timeout_secint否3查询 InfluxDB 的超时时间（秒）。
common-optionsconfig否-Source 插件通用参数，详见 Source 通用选项。

url [string]​

连接到 InfluxDB 的 URL，例如 http://influxdb-host:8086。

sql [string]​

用于读取数据的 InfluxQL 查询，例如：

select name, age from test

schema [config]​

上游数据的 schema 信息，更多语法参考 Schema 特性。 例如：

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

database [string]​

InfluxDB 数据库名称。

username [string]​

InfluxDB 用户名。必须和 password 一起配置。

password [string]​

InfluxDB 密码。必须和 username 一起配置。

split_column [string]​

用于把一次查询切分成多个范围查询的整数列。启用并行范围读取时必须配置。

提示：

  • InfluxDB tags 不支持作为分割主键，因为 tags 的类型只能是字符串。
  • InfluxDB time 不支持作为分割主键，因为 time 字段无法参与数学计算。
  • 目前，split_column 仅支持整数数据分割，不支持 float、string、date 等类型。
  • split_column、lower_bound、upper_bound、partition_num 需要一起配置。
  • 如果切分读取的 SQL 中包含过滤条件，请在 sql 里使用小写 where，例如 select * from test where age > 0。当前切分解析逻辑区分大小写。
  • where 是配置校验中保留的选项，但当前切分逻辑会从 sql 里读取过滤条件。请把过滤条件写在 sql 中，不要单独配置 where。

upper_bound [int]​

split_column 列的上界。启用并行范围读取时使用。

lower_bound [int]​

split_column 列的下界。启用并行范围读取时使用。

连接器会把 split_column 范围切成 partition_num 份；若 partition_num = 1，则使用整段范围；若 partition_num 小于 upper_bound - lower_bound，则按 upper_bound - lower_bound 切分。

例如 lower_bound = 1、upper_bound = 10、partition_num = 2、 sql = "select * from test where age > 0 and age < 10" 会被切分成：

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]​

查询切分数量。须与 lower_bound、upper_bound、split_column 一起配置。

提示：确保 upper_bound - lower_bound 能被 partition_num 整除，否则查询结果会重叠。

epoch [string]​

InfluxDB 返回的时间精度。可选值：H、m、s、MS、u、n，默认值为 n。

query_timeout_sec [int]​

InfluxDB 客户端的查询超时时间，单位为秒。

connect_timeout_ms [long]​

连接到 InfluxDB 的超时时间，单位为毫秒。

通用选项​

Source 插件通用参数，请参考 Source 通用选项 详见。

任务示例​

使用并行范围读取​

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

不使用并行范围读取​

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

使用 InfluxQL 时区查询​

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

变更日志​

Change Log
ChangeCommitVersion
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] influxdb options (#8966)https://github.com/apache/seatunnel/commit/9f498b81332.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Improve] Improve some connectors prepare check error message (#7465)https://github.com/apache/seatunnel/commit/6930a25edd2.3.8
[Improve][Connector] Add multi-table sink option check (#7360)https://github.com/apache/seatunnel/commit/2489f6446b2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)https://github.com/apache/seatunnel/commit/c4ca74122c2.3.6
Support multi-table sink feature for influxdb (#6278)https://github.com/apache/seatunnel/commit/56f13e920d2.3.5
[Improve][Zeta] Add classloader cache mode to fix metaspace leak (#6355)https://github.com/apache/seatunnel/commit/9c3c2f183d2.3.5
[Test][E2E] Add thread leak check for connector (#5773)https://github.com/apache/seatunnel/commit/1f2f3fc5f02.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/deed9c62c32.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Improve][Connector-V2] Remove scheduler in InfluxDB sink (#5271)https://github.com/apache/seatunnel/commit/f459f500cb2.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)https://github.com/apache/seatunnel/commit/51c0d709ba2.3.4
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Improve][SourceConnector] Unifie InfluxDB source fields to schema (#3897)https://github.com/apache/seatunnel/commit/85a984a64f2.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API &amp; Connector &amp; Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Improve][Connector-V2][Influxdb] Unified exception for influxdb source & sink connector (#3558)https://github.com/apache/seatunnel/commit/4686f35d682.3.0
[Feature][Connector][influx] Expose configurable options in influx db (#3392)https://github.com/apache/seatunnel/commit/b247ff0aef2.3.0
[Feature][Connector-V2] influxdb sink connector (#3174)https://github.com/apache/seatunnel/commit/630e8847912.3.0
[Feature][Connector-V2] Add influxDB connector source (#2697)https://github.com/apache/seatunnel/commit/1d70ea30842.3.0-beta