下载安装

下载地址
下载后在解压目录中，输入cmd执行exe文件
浏览器访问localhost:8086
选择快速开始，填写用户信息，组织信息

数据操作

Line Protocol

选择Enter Manually执行语句

语法

InfluxDB使用行协议写入数据点。它是一种基于文本的格式，提供数据点的度量、标记集、字段集和时间戳。
measurementName,tagKey=tagValue fieldKey="fieldValue" 1465839830100400200
--------------- --------------- --------------------- -------------------
| | | |
Measurement Tag set Field set Timestamp
例:
myMeasurement,tag1=value1,tag2=value2 fieldKey="fieldValue" 1556813561098000000
由换行符分隔的行 \ n表示InfluxDB中的单个点。线路协议对空格敏感。

Explore

可进行页面上的筛选，点击script Editor查看执行的语句

可切换数据呈现的样式

查询数据

声明数据库：from(bucket:"example-bucket")
指定查询范围：|> range(start: -1h)
设置筛选条件：|> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage_system" and r.cpu == "cpu-total")
输出结果：yield()
Flux自动假定在每个脚本的末尾有一个yield()函数，用于输出和可视化数据。只有在同一个Flux查询中包含多个查询时，才需要显式地调用yield()。每一组返回的数据都需要使用yield()函数命名。

完整语句：

from(bucket: "example-bucket")

    |> range(start: -15m)

    |> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage_system" and r.cpu == "cpu-total")

    |> yield(name: "test")

java开发

引入依赖

<dependency>

  <groupId>com.influxdb</groupId>

  <artifactId>influxdb-client-java</artifactId>

  <version>3.1.0</version>

</dependency>

数据模型

@Data

@Accessors(chain = true)

@Measurement(name = "monitoring_data")

public class MonData {

  @Column(tag = true)

  private String pointName;

  @Column(tag = true)

  private String indexName;

  @Column private Double value;

  @Column(timestamp = true)

  private Instant time;

}

@InfluxColumn为自定义注解，用于拼接查询语句构造map函数使用

@Retention(RetentionPolicy.RUNTIME)

@Target(ElementType.FIELD)

public @interface InfluxColumn {

  String value();

}

@Data

public class MonDataDTO {

  // tag1名称

  @InfluxColumn("pointName")

  private String pointName;

  // tag2名称

  @InfluxColumn("indexName")

  private String indexName;

  // 时间片开始时间

  @InfluxColumn("_start")

  private String start;

  // 时间片结束时间

  @InfluxColumn("_stop")

  private String stop;

  // 数据产生时间

  @InfluxColumn("_time")

  private String time;

  // 值

  @InfluxColumn("_value")

  private String value;

}

@Data

public class SearchParams<T> {

  // 查询时间范围开始时间

  private String start;

  // 时间戳字段排序规则，true：降序

  private Boolean sortRule = true;

  // 查询时间范围结束时间

  private String end;

  // 时间间隔

  private String every;

  // 筛选条件

  private List<String> filterList;

  // map构造的目标类对象

  private Class<? extends T> mapClazz;

}

功能类

@Repository

@Slf4j

public class InfluxRepository {

  @Autowired private WriteApi writeApi;

  @Autowired private QueryApi queryApi;

  @Autowired private InfluxdbConfigProp influxdbConfigProp;

  /**

   * 向influx写入数据

   *

   * @param data 写入数据实体

   */

  public <T> void writeData(T data) {

    writeApi.writeMeasurement(

        influxdbConfigProp.getBucket(), influxdbConfigProp.getOrg(), WritePrecision.MS, data);

  }

  /**

   * 查询数据

   *

   * @param params 查询参数

   */

  public <T> List<FluxTable> findMonitoringData(SearchParams<T> params) {

    StringBuffer queryBuffer = new StringBuffer();

    // BUCKET

    queryBuffer.append("from(bucket: \"");

    queryBuffer.append(influxdbConfigProp.getBucket());

    // 时间范围条件

    queryBuffer.append("\") \n|> range(start: ");

    queryBuffer.append(params.getStart());

    queryBuffer.append(", stop: ");

    queryBuffer.append(params.getEnd());

    queryBuffer.append(")\n");

    List<String> filterList = params.getFilterList();

    if (!CollectionUtils.isEmpty(filterList)) {

      queryBuffer.append("  |> filter(fn: (r) => ");

      // 拼接查询条件

      for (int i = 0; i < filterList.size(); i++) {

        String[] filters = filterList.get(i).split(">");

        queryBuffer.append("r[\"");

        queryBuffer.append(filters[0]);

        queryBuffer.append("\"]");

        queryBuffer.append(filters[1]);

        if (i < filterList.size() - 1) queryBuffer.append(" and ");

      }

      queryBuffer.append(")\n");

    }

    // aggregateWindow函数

    queryBuffer.append("  |> aggregateWindow(every: ");

    queryBuffer.append(params.getEvery());

    queryBuffer.append(",fn: first, createEmpty: true)\n");

    // 为查询结果添加排序

    queryBuffer.append("  |> sort(columns: [\"_time\"], desc: ");

    queryBuffer.append(params.getSortRule().booleanValue());

    queryBuffer.append(")\n");

    // map函数语句拼接

    Class<? extends T> mapClazz = params.getMapClazz();

    if (!ObjectUtils.isEmpty(mapClazz)) {

      queryBuffer.append("  |> map(");

      queryBuffer.append(" fn:(r) => { \n");

      queryBuffer.append("    return {\n");

      Field[] fields = mapClazz.getDeclaredFields();

      // 目标实体字段和influx查询结果字段的映射

      Map<String, String> fieldMap = new HashMap<>();

      for (Field field : fields) {

        InfluxColumn influxColumn = field.getAnnotation(InfluxColumn.class);

        if (influxColumn != null) {

          fieldMap.put(field.getName(), influxColumn.value());

        }

      }

      // 若有需要映射的字段则构建语句

      if (!CollectionUtils.isEmpty(fieldMap)) {

        for (String key : fieldMap.keySet()) {

          queryBuffer.append(key);

          queryBuffer.append(": r[\"");

          queryBuffer.append(fieldMap.get(key));

          queryBuffer.append("\"],\n");

        }

        queryBuffer.append("}})\n");

      }

    }

    String influxQl = queryBuffer.toString();

    log.info("查询语句, {}", influxQl);

    List<FluxTable> queryData = queryApi.query(influxQl, influxdbConfigProp.getOrg());

    return queryData;

  }

}

@Service

@Slf4j

public class InfluxQueryService {

  @Autowired private ObjectMapper objectMapper;

  @Autowired private InfluxRepository influxRepository;

  /**

   * 监测数据查询

   *

   * @param start 起始范围时间点

   * @param end 结束范围时间点

   * @param every 时间片

   * @param filterList 筛选条件集合（集合内元素例：pointName>csd-001）

   * @param clazz 去除数据时map对象映射的类对象

   * @param sort 时间字段排序规则

   */

  public <T> List<T> findMonitoringDataInFluxDB(

      String start,

      String end,

      String every,

      List<String> filterList,

      Class<? extends T> clazz,

      boolean sort) {

    // mainTag和 subTag需要特殊处理，将逗号替换成"|"正则表达

    filterList =

        filterList.stream()

            .map(filter -> StringUtils.replace(filter, ",", "|"))

            .collect(Collectors.toList());

    SearchParams<T> searchParams = new SearchParams<>();

    searchParams.setStart(start);

    searchParams.setEnd(end);

    searchParams.setEvery(every);

    searchParams.setFilterList(filterList);

    searchParams.setMapClazz(clazz);

    searchParams.setSortRule(sort);

    List<FluxTable> fluxTableList = influxRepository.findMonitoringData(searchParams);

    return mapFluxData(fluxTableList, clazz);

  }

  /**

   * 解析原始数据

   *

   * @param data 原始数据

   */

  public <T> List<T> mapFluxData(List<FluxTable> data, Class<? extends T> clazz) {

    List<T> result = new LinkedList<>();

    for (FluxTable ft : data) {

      List<FluxRecord> records = ft.getRecords();

      for (FluxRecord rc : records) {

        try {

          T originData =

              objectMapper.readValue(objectMapper.writeValueAsString(rc.getValues()), clazz);

          result.add(originData);

        } catch (JsonProcessingException e) {

          log.error("influx查询数据转换为DTO时解析出错");

          throw new RuntimeException(e);

        }

      }

    }

    return result;

  }

}

业务Service构造查询条件，并提供相应的：查询结果实体 => 实体之间的转换方法

/**

   * 设备指标监测值

   *

   * @param start 起始范围时间点

   * @param end 结束范围时间点

   * @param every 时间片

   * @param tagName 设备id

   */

  public List<MonDataDTO> getMonitoringData(

      String start, String end, String every, String tagName) {

    // 筛选条件

    List<String> filterList = new ArrayList<>();

    filterList.add("_measurement> == \"monitoring_data\"");

    filterList.add("tagName> =~/" + tagName + "/");

    // 处理时间参数

    String startDate;

    String endDate;

    LocalDate startLocalDate = LocalDate.parse(start).plusDays(-1);

    // 一天内的数据 (开始时间的前一天的23点，到结束时间的23点，时区原因查询时时间减去8小时)

    // 跨天的数据（开始和结束时间减8小时）

    String endTime = (start.equals(end) ? "T15:00:00Z" : "T16:00:00Z");

    startDate = startLocalDate + endTime;

    endDate = end + endTime;

    List<MonDataDTO> dataInFluxDB =

        influxQueryService.findMonitoringDataInFluxDB(

            startDate, endDate, every, filterList, MonDataDTO.class, false);

    return dataInFluxDB;

  }

配置类

@Data

@ConfigurationProperties(prefix = "influxdb")

@Component

public class InfluxdbConfigProp {

  private String token;

  private String bucket;

  private String org;

  private String url;

}



@Configuration

public class InfluxdbConfig {

  @Autowired private InfluxdbConfigProp influxdbConfigProp;

  @Bean

  public InfluxDBClient influxDBClient() {

    InfluxDBClient influxClient =

        InfluxDBClientFactory.create(

            influxdbConfigProp.getUrl(), influxdbConfigProp.getToken().toCharArray());

    influxClient.setLogLevel(LogLevel.BASIC);

    return influxClient;

  }

  @Bean

  public WriteApi writeApi(InfluxDBClient influxDBClient) {

    WriteOptions writeOptions =

        WriteOptions.builder()

            .batchSize(5000)

            .flushInterval(1000)

            .bufferLimit(10000)

            .jitterInterval(1000)

            .retryInterval(5000)

            .build();

    return influxDBClient.getWriteApi(writeOptions);

  }

  @Bean

  public QueryApi queryApi(InfluxDBClient influxDBClient) {

    return influxDBClient.getQueryApi();

  }

}

常用函数

window()

使用window()函数根据时间界限对数据进行分组。window()传递的最常用参数是every，它定义了窗口之间的持续时间。也可以使用其他参数，但是对于本例，将基本数据集窗口化为一分钟窗口。
dataSet |> window(every: 1m)

first()和last()

获取查询结果的第一条或最后一条

drop()

删除查询结果的指定列
|> drop(columns: ["host"])

sort()和limit()

排序和分页
|> sort(columns: ["index", "time"], desc: true)
|> limit(n: 10)
n参数为pageSize

timedMovingAverage()

对于表中的每一行，timedMovingAverage()返回当前值和上一个周期(持续时间)中所有行值的平均值。它以每个参数定义的频率返回移动平均线。
|> timedMovingAverage(every: 1h, period: 1h)

aggregateWindow()

|> aggregateWindow(every: 1h, fn: first, createEmpty: true)
每一小时时间片的第一条记录，空数据以null填充

map()

|> map( fn:(r) => {

    return {

      code: r["code"],

      time: r["_time"],

      value: r["_value"],

      index: r["indexName"]

      }

   }

  )

注意事项

tag与tag之间用逗号分隔
field与field之间用逗号分隔
tag与field之间用空格分隔
tag都是string类型，不需要引号将value包裹
tag的值不能有空格
写入数据时，若tag和时间戳都相同的多条记录，则最后只会保存一条

巴特西

influxDB2.2