Skip to content

InfluxDB Writer

InfluxDB Writer 插件实现了将数据写入 InfluxDB 读取数据的功能。 底层实现上，是通过调用 InfluQL 语言接口，构建插入语句，然后进行数据插入。

示例

以下示例用来演示该插件从内存读取数据并写入到指定表

创建需要的库

通过以下命令来创建需要写入的库

# create database
influx --execute "CREATE DATABASE addax"

创建 job 文件

创建 job/stream2influxdb.json 文件，内容如下：

{
  "job": {
    "setting": {
      "speed": {
        "channel": 1,
        "bytes": -1
      }
    },
    "content": {
      "reader": {
        "name": "streamreader",
        "parameter": {
          "column": [
            {
              "random": "2001-01-01 00:00:00, 2016-07-07 23:59:59",
              "type": "date"
            },
            {
              "random": "1,1000",
              "type": "long"
            },
            {
              "random": "1,10",
              "type": "string"
            },
            {
              "random": "1000,50000",
              "type": "double"
            }
          ],
          "sliceRecordCount": 10
        }
      },
      "writer": {
        "name": "influxdbwriter",
        "parameter": {
          "connection": {
            "endpoint": "http://localhost:8086",
            "database": "addax",
            "table": "addax_tbl"
          },
          "connTimeout": 15,
          "readTimeout": 20,
          "writeTimeout": 20,
          "username": "influx",
          "password": "influx123",
          "column": [
            {
              "name": "time",
              "type": "timestamp"
            },
            {
              "name": "user_id",
              "type": "int"
            },
            {
              "name": "user_name",
              "type": "string"
            },
            {
              "name": "salary",
              "type": "double"
            }
          ],
          "preSql": [
            "delete from addax_tbl"
          ],
          "batchSize": 1024,
          "retentionPolicy": {
            "name": "one_day_only",
            "duration": "1d",
            "replication": 1
          }
        }
      }
    }
  }
}

运行

执行下面的命令进行数据采集

bin/addax.sh job/stream2influxdb.json

参数说明

配置项 是否必须 数据类型 默认值 描述
endpoint 是 string 无 InfluxDB 连接串
username 是 string 无 数据源的用户名
password 否 string 无 数据源指定用户名的密码
database 是 string 无 数据源指定的数据库
table 是 string 无 要写入的表（指标）
column 是 list 无 所配置的表中需要同步的列名集合
connTimeout 否 int 15 设置连接超时值，单位为秒
readTimeout 否 int 20 设置读取超时值，单位为秒
writeTimeout 否 int 20 设置写入超时值，单位为秒
preSql 否 list 无 插入数据前执行的SQL语句
postSql 否 list 无 数据插入完毕后需要执行的语句
retentionPolicy 否 map 无 设置数据库的 Retention Policy 策略

column

InfluxDB 作为时序数据库，需要每条记录都有时间戳字段，因此这里会把 column 配置的第一个字段默认当作时间戳

retentionPolicy

设定数据库的 Retention Policy 策略，依据给定的配置，在指定数据库上创建一条 Retention Policy 信息。 有关 Retention Policy 更详细的信息，可以参考 [官方文档][2]

类型转换

当前支持 InfluxDB 的基本类型

限制

  1. 当前插件仅支持 1.x 版本，2.0 及以上并不支持