如何使用PyFlink将消息下沉写入InfluxDB

一、InfluxDB连接器Docker容器部署方法 #

Flink官方未内置InfluxDB SQL连接器，使用Apache Bahir提供的适配连接器即可，操作步骤如下：

• 先确认当前PyFlink环境的FLINK_VERSION取值，必须下载和Flink主版本完全一致的连接器jar包，否则会出现类冲突、加载失败的问题。
• 修改项目路径下的Dockerfile，找到原有下载连接器jar包的RUN指令块，删除原Elasticsearch连接器的下载行，新增InfluxDB连接器的下载命令，参考修改后的配置如下：

# Download connector libraries
RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/${FLINK_VERSION}/flink-json-${FLINK_VERSION}.jar; \
    wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.12/${FLINK_VERSION}/flink-sql-connector-kafka_2.12-${FLINK_VERSION}.jar; \
    # 下载InfluxDB 2.x连接器，若使用1.x版本将包名中influxdb2替换为influxdb即可
    wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/bahir/flink-connector-influxdb2_2.12/${FLINK_VERSION}/flink-connector-influxdb2_2.12-${FLINK_VERSION}.jar;

• 重新执行Docker镜像构建命令，待镜像构建完成后启动容器，进入容器内/opt/flink/lib目录，确认InfluxDB连接器jar包已存在、文件权限和其他连接器包一致即部署完成。
• 额外检查项：确认编排文件中已加入InfluxDB服务，且和Flink服务处于同一Docker网络下，保证Flink可正常访问InfluxDB默认8086端口。

二、InfluxDB Sink建表语句修改 #

InfluxDB是时序数据库，建表时需要明确指定标签（tag，带索引的维度字段）、存储值（field，指标字段）、时间戳字段三类核心要素，原Elasticsearch的建表语句修改后参考如下（以InfluxDB 2.x为例）：

CREATE TABLE influx_sink (
    id VARCHAR,
    value DOUBLE,
    -- 时序数据必需的时间戳字段
    ts TIMESTAMP(3),
    -- 从上游携带时间的话配置水位线，也可配置连接器自动生成写入时间
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
    'connector' = 'influxdb2', -- 使用InfluxDB 1.x时此处值改为'influxdb'
    'url' = 'http://influxdb:8086', -- Docker网络内InfluxDB服务地址
    'bucket' = 'platform_metrics', -- 2.x版本对应存储bucket，1.x版本替换为'database'参数指定库名
    'org' = 'your_org', -- 2.x版本配置组织名，1.x版本无需该参数
    'token' = 'your_access_token', -- 2.x版本配置访问令牌，1.x版本替换为'username'、'password'参数做鉴权
    'measurement' = 'platform_measurements_1', -- 对应原ES的index，即时序测量名
    'tag' = 'id', -- 指定作为tag的维度字段，多个tag用分号分隔
    'field' = 'value', -- 指定作为存储指标的field字段
    'timestamp' = 'ts', -- 指定作为时序点时间戳的字段
    'write.buffer-size' = '1000', -- 可选：写入缓冲区大小
    'write.flush-interval' = '1s' -- 可选：数据刷写间隔
);

配置注意事项 #

• 不要混淆tag和field配置：tag是带索引的维度字段（比如示例中的设备id），适合做查询过滤条件；field是无索引的指标值（比如示例中的采集数值），配置错误会导致查询性能大幅下降。
• 如果业务不需要自定义事件时间，可以不在表结构中显式定义ts字段，配置连接器自动生成写入时间即可。
• 若使用InfluxDB 1.x版本，删除WITH参数中bucket、org、token配置项，替换为对应1.x版本的database、username、password参数即可。

内容的提问来源于stack exchange，提问作者Ermolai