InfluxDB2的数据查询示例

有用influxdb2 不支持sql，并且实质是个列存储数据库，这里基于

influxdb-client-java 和 beanutils反射，写了个数据查询，把结果以行对象的形式返回的工具类。

package com.joy.malltools.influxdb2;import com.influxdb.client.QueryApi;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import org.apache.commons.beanutils.PropertyUtils;import java.util.List;
import java.util.Map;/*** 对应influxDB2的查询结果的处理*/
public class InfluxDB2Util2 {/**** @param clz 类型* @param queryApi 待转换的查询集合* @param sql influxdb2的查询语句* @param extStrColumns 额外的String类型的字段名，一般为tag* @return clz类型的对象数组* @param <T>*/public static <T> T[] testStatSql(Class<T> clz, QueryApi queryApi, String sql, String[] extStrColumns) {System.out.println(sql);List<FluxTable> query = queryApi.query(sql);int count = query.get(0).getRecords().size();System.out.println("记录条数.size " + count);T[] vos = (T[]) java.lang.reflect.Array.newInstance(clz, count);for (int k = 0; k < count; k++) {try {vos[k] = clz.newInstance();} catch (Exception e) {e.printStackTrace();}}for (int i=0;i<query.size();i++) {FluxTable fluxTable = query.get(i);System.out.println("records.size " + fluxTable.getRecords().size());List<FluxRecord> recordList = fluxTable.getRecords();for (int j=0;j<recordList.size();j++) {FluxRecord fr = recordList.get(j);System.out.println("时间  " + fr.getTime() + "  求记录数 " + fr.getValue());System.out.println("时间  " + fr.getTime() + "  字段名 " + fr.getField() +  " measurement " + fr.getMeasurement());Map<String, Object> valueMap = fr.getValues();System.out.print("key : " + valueMap.get("_field") + " ; value " + valueMap.get("_value") + "|");try {PropertyUtils.setProperty(vos[j], valueMap.get("_field")+"", valueMap.get("_value"));} catch (Exception e) {e.printStackTrace();}if (extStrColumns != null && extStrColumns.length > 0)for (String colStr : extStrColumns) {setProperty(vos[j], colStr, fr.getValueByKey(colStr)+"");}
//                setProperty(vos[j], "location", fr.getValueByKey("location")+"");
//                setProperty(vos[j], "description", fr.getValueByKey("description")+"");}System.out.println("");}System.out.println("数据集合 --> ");for (Object vo: vos) {System.out.println("vo --> " + vo);}return vos;}private static void setProperty(Object obj, String fieldName, String value) {try {PropertyUtils.setProperty(obj, fieldName, value);} catch (Exception e) {e.printStackTrace();}}}

主要依赖的jar

 <dependency><groupId>com.influxdb</groupId><artifactId>influxdb-client-java</artifactId><version>6.5.0</version></dependency><dependency><groupId>commons-beanutils</groupId><artifactId>commons-beanutils</artifactId><version>1.9.4</version></dependency>

调用示例

String sql = "from(bucket: \"test\")\n" +"  |> range(start: -3d)\n" +"  |> filter(fn: (r) => r[\"_measurement\"] == \"h2o_feet\" )\n" +"  |> yield(name: \"mean\")";InfluxDBClient influxDBClient = InfluxDBClientFactory.create(url,token.toCharArray(),org,bucket);QueryApi queryApi = influxDBClient.getQueryApi();String[] strColAry = new String[]{"location", "description"};WaterVO[] objs = InfluxDB2Util2.testStatSql(WaterVO.class, queryApi, sql, strColAry);System.out.println("<-- ---------------- vo ----------------  --> ");for (WaterVO vo: objs) {System.out.println("vo --> "+vo);}