查询数据
主要查询功能
TDengine 采用 SQL 作为查询语言。应用程序可以通过 REST API 或连接器发送 SQL 语句,用户还可以通过 TDengine 命令行工具 taos 手动执行 SQL 即席查询(Ad-Hoc Query)。TDengine 支持如下查询功能:
- 单列、多列数据查询
- 标签和数值的多种过滤条件:>, <, =, <>, like 等
- 聚合结果的分组(Group by)、排序(Order by)、约束输出(Limit/Offset)
- 时间窗口(Interval)、会话窗口(Session)和状态窗口(State_window)等窗口切分聚合查询
- 数值列及聚合结果的四则运算
- 时间戳对齐的连接查询(Join Query: 隐式连接)操作
- 多种聚合/计算函数: count, max, min, avg, sum, twa, stddev, leastsquares, top, bottom, first, last, percentile, apercentile, last_row, spread, diff 等
例如:在命令行工具 taos 中,从表 d1001 中查询出 voltage > 215 的记录,按时间降序排列,仅仅输出 2 条。
taos> select * from d1001 where voltage > 215 order by ts desc limit 2;
ts | current | voltage | phase |
======================================================================================
2018-10-03 14:38:16.800 | 12.30000 | 221 | 0.31000 |
2018-10-03 14:38:15.000 | 12.60000 | 218 | 0.33000 |
Query OK, 2 row(s) in set (0.001100s)
为满足物联网场景的需求,TDengine 支持几个特殊的函数,比如 twa(时间加权平均),spread (最大值与最小值的差),last_row(最后一条记录)等,更多与物联网场景相关的函数将添加进来。
具体的查询语法请看 TDengine SQL 的数据查询 章节。
多表聚合查询
物联网场景中,往往同一个类型的数据采集点有多个。TDengine 采用超级表(STable)的概念来描述某一个类型的数据采集点,一张普通的表来描述一个具体的数据采集点。同时 TDengine 使用标签来描述数据采集点的静态属性,一个具体的数据采集点有具体的标签值。通过指定标签的过滤条件,TDengine 提供了一高效的方法将超级表(某一类型的数据采集点)所属的子表进行聚合查询。对普通表的聚合函数以及绝大部分操作都适用于超级表,语法完全一样。
示例一
在 TDengine CLI,查找加利福尼亚州所有智能电表采集的电压平均值,并按照 location 分组。
taos> SELECT AVG(voltage), location FROM meters GROUP BY location;
avg(voltage) | location |
===============================================================================================
219.200000000 | California.SanFrancisco |
221.666666667 | California.LosAngeles |
Query OK, 2 rows in database (0.005995s)
示例二
在 TDengine CLI, 查找 groupId 为 2 的所有智能电表的记录条数,电流的最大值。
taos> SELECT count(*), max(current) FROM meters where groupId = 2;
cunt(*) | max(current) |
==================================
5 | 13.4 |
Query OK, 1 row(s) in set (0.002136s)
在 TDengine SQL 的数据查询 一章,查询类操作都会注明是否支持超级表。
降采样查询、插值
物联网场景里,经常需要通过降采样(down sampling)将采集的数据按时间段进行聚合。TDengine 提供了一个简便的关键词 interval 让按照时间窗口的查询操作变得极为简单。比如,将智能电表 d1001 采集的电流值每 10 秒钟求和
taos> SELECT _wstart, sum(current) FROM d1001 INTERVAL(10s);
_wstart | sum(current) |
======================================================
2018-10-03 14:38:00.000 | 10.300000191 |
2018-10-03 14:38:10.000 | 24.900000572 |
Query OK, 2 rows in database (0.003139s)
降采样操作也适用于超级表,比如:将加利福尼亚州所有智能电表采集的电流值每秒钟求和
taos> SELECT _wstart, SUM(current) FROM meters where location like "California%" INTERVAL(1s);
_wstart | sum(current) |
======================================================
2018-10-03 14:38:04.000 | 10.199999809 |
2018-10-03 14:38:05.000 | 23.699999809 |
2018-10-03 14:38:06.000 | 11.500000000 |
2018-10-03 14:38:15.000 | 12.600000381 |
2018-10-03 14:38:16.000 | 34.400000572 |
Query OK, 5 rows in database (0.007413s)
降采样操作也支持时间偏移,比如:将所有智能电表采集的电流值每秒钟求和,但要求每个时间窗口从 500 毫秒开始
taos> SELECT _wstart, SUM(current) FROM meters INTERVAL(1s, 500a);
_wstart | sum(current) |
======================================================
2018-10-03 14:38:03.500 | 10.199999809 |
2018-10-03 14:38:04.500 | 10.300000191 |
2018-10-03 14:38:05.500 | 13.399999619 |
2018-10-03 14:38:06.500 | 11.500000000 |
2018-10-03 14:38:14.500 | 12.600000381 |
2018-10-03 14:38:16.500 | 34.400000572 |
Query OK, 6 rows in database (0.005515s)
物联网场景里,每个数据采集点采集数据的时间是难同步的,但很多分析算法(比如 FFT)需要把采集的数据严格按照时间等间隔的对齐,在很多系统里,需要应用自己写程序来处理,但使用 TDengine 的降采样操作就轻松解决。
如果一个时间间隔里,没有采集的数据,TDengine 还提供插值计算的功能。
语法规则细节请见 TDengine SQL 的按时间窗口切分聚合 章节。
示例代码
查询数据
在 SQL 写入 一章,我们创建了 power 数据库,并向 meters 表写入了一些数据,以下示例代码展示如何查询这个表的数据。
- Java
- Python
- Go
- Rust
- Node.js
- C#
- C
- PHP
package com.taos.example;
import java.sql.*;
public class RestQueryExample {
private static Connection getConnection() throws SQLException {
String jdbcUrl = "jdbc:TAOS-RS://localhost:6041/power?user=root&password=taosdata";
return DriverManager.getConnection(jdbcUrl);
}
private static void printRow(ResultSet rs) throws SQLException {
ResultSetMetaData meta = rs.getMetaData();
for (int i = 1; i <= meta.getColumnCount(); i++) {
String value = rs.getString(i);
System.out.print(value);
System.out.print("\t");
}
System.out.println();
}
private static void printColName(ResultSet rs) throws SQLException {
ResultSetMetaData meta = rs.getMetaData();
for (int i = 1; i <= meta.getColumnCount(); i++) {
String colLabel = meta.getColumnLabel(i);
System.out.print(colLabel);
System.out.print("\t");
}
System.out.println();
}
private static void processResult(ResultSet rs) throws SQLException {
printColName(rs);
while (rs.next()) {
printRow(rs);
}
}
private static void queryData() throws SQLException {
try (Connection conn = getConnection()) {
try (Statement stmt = conn.createStatement()) {
ResultSet rs = stmt.executeQuery("SELECT AVG(voltage) FROM meters GROUP BY location");
processResult(rs);
}
}
}
public static void main(String[] args) throws SQLException {
queryData();
}
}
// possible output:
// avg(voltage) location
// 222.0 California.LosAngeles
// 219.0 California.SanFrancisco
通过迭代逐行获取查询结果。
def query_api_demo(conn: taos.TaosConnection):
result: taos.TaosResult = conn.query("SELECT tbname, * FROM meters LIMIT 2")
print("field count:", result.field_count)
print("meta of fields[1]:", result.fields[1])
print("======================Iterate on result=========================")
for row in result:
print(row)
# field count: 7
# meta of fields[1]: {name: ts, type: 9, bytes: 8}
# ======================Iterate on result=========================
# ('d1003', datetime.datetime(2018, 10, 3, 14, 38, 5, 500000), 11.800000190734863, 221, 0.2800000011920929, 'california.losangeles', 2)
# ('d1003', datetime.datetime(2018, 10, 3, 14, 38, 16, 600000), 13.399999618530273, 223, 0.28999999165534973, 'california.losangeles', 2)
一次获取所有查询结果,并把每一行转化为一个字典返回。
def fetch_all_demo(conn: taos.TaosConnection):
result: taos.TaosResult = conn.query("SELECT ts, current FROM meters LIMIT 2")
rows = result.fetch_all_into_dict()
print("row count:", result.row_count)
print("===============all data===================")
print(rows)
# row count: 2
# ===============all data===================
# [{'ts': datetime.datetime(2018, 10, 3, 14, 38, 5, 500000), 'current': 11.800000190734863},
# {'ts': datetime.datetime(2018, 10, 3, 14, 38, 16, 600000), 'current': 13.399999618530273}]
package main
import (
"database/sql"
"log"
"time"
_ "github.com/taosdata/driver-go/v3/taosRestful"
)
func main() {
var taosDSN = "root:taosdata@http(localhost:6041)/power"
taos, err := sql.Open("taosRestful", taosDSN)
if err != nil {
log.Fatalln("failed to connect TDengine, err:", err)
}
defer taos.Close()
rows, err := taos.Query("SELECT ts, current FROM meters LIMIT 2")
if err != nil {
log.Fatalln("failed to select from table, err:", err)
}
defer rows.Close()
for rows.Next() {
var r struct {
ts time.Time
current float32
}
err := rows.Scan(&r.ts, &r.current)
if err != nil {
log.Fatalln("scan error:\n", err)
return
}
log.Println(r.ts, r.current)
}
}
use taos::sync::*;
fn main() -> anyhow::Result<()> {
let taos = TaosBuilder::from_dsn("ws:///power")?.build()?;
let mut result = taos.query("SELECT ts, current FROM meters LIMIT 2")?;
// print column names
let meta = result.fields();
println!("{}", meta.iter().map(|field| field.name()).join("\t"));
// print rows
let rows = result.rows();
for row in rows {
let row = row?;
for (_name, value) in row {
print!("{}\t", value);
}
println!();
}
Ok(())
}
// output(suppose you are in +8 timezone):
// ts current
// 2018-10-03T14:38:05+08:00 10.3
// 2018-10-03T14:38:15+08:00 12.6
const taos = require("@tdengine/client");
const conn = taos.connect({ host: "localhost", database: "power" });
const cursor = conn.cursor();
const query = cursor.query("SELECT ts, current FROM meters LIMIT 2");
query.execute().then(function (result) {
result.pretty();
});
// output:
// Successfully connected to TDengine
// ts | current |
// =======================================================
// 2018-10-03 14:38:05.000 | 10.3 |
// 2018-10-03 14:38:15.000 | 12.6 |
using TDengineDriver;
using TDengineDriver.Impl;
using System.Runtime.InteropServices;
namespace TDengineExample
{
internal class QueryExample
{
static void Main()
{
IntPtr conn = GetConnection();
try
{
// run query
IntPtr res = TDengine.Query(conn, "SELECT * FROM meters LIMIT 2");
if (TDengine.ErrorNo(res) != 0)
{
throw new Exception("Failed to query since: " + TDengine.Error(res));
}
// get filed count
int fieldCount = TDengine.FieldCount(res);
Console.WriteLine("fieldCount=" + fieldCount);
// print column names
List<TDengineMeta> metas = LibTaos.GetMeta(res);
for (int i = 0; i < metas.Count; i++)
{
Console.Write(metas[i].name + "\t");
}
Console.WriteLine();
// print values
List<Object> resData = LibTaos.GetData(res);
for (int i = 0; i < resData.Count; i++)
{
Console.Write($"|{resData[i].ToString()} \t");
if (((i + 1) % metas.Count == 0))
{
Console.WriteLine("");
}
}
Console.WriteLine();
// Free result after use
TDengine.FreeResult(res);
}
finally
{
TDengine.Close(conn);
}
}
static IntPtr GetConnection()
{
string host = "localhost";
short port = 6030;
string username = "root";
string password = "taosdata";
string dbname = "power";
var conn = TDengine.Connect(host, username, password, dbname, port);
if (conn == IntPtr.Zero)
{
throw new Exception("Connect to TDengine failed");
}
else
{
Console.WriteLine("Connect to TDengine success");
}
return conn;
}
}
}
// output:
// Connect to TDengine success
// fieldCount=6
// ts current voltage phase location groupid
// 1648432611249 10.3 219 0.31 California.SanFrancisco 2
// 1648432611749 12.6 218 0.33 California.SanFrancisco 2
// compile with:
// gcc -o query_example query_example.c -ltaos
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <taos.h>
typedef int16_t VarDataLenT;
#define TSDB_NCHAR_SIZE sizeof(int32_t)
#define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
#define GET_FLOAT_VAL(x) (*(float *)(x))
#define GET_DOUBLE_VAL(x) (*(double *)(x))
#define varDataLen(v) ((VarDataLenT *)(v))[0]
int printRow(char *str, TAOS_ROW row, TAOS_FIELD *fields, int numFields) {
int len = 0;
char split = ' ';
for (int i = 0; i < numFields; ++i) {
if (i > 0) {
str[len++] = split;
}
if (row[i] == NULL) {
len += sprintf(str + len, "%s", "NULL");
continue;
}
switch (fields[i].type) {
case TSDB_DATA_TYPE_TINYINT:
len += sprintf(str + len, "%d", *((int8_t *)row[i]));
break;
case TSDB_DATA_TYPE_UTINYINT:
len += sprintf(str + len, "%u", *((uint8_t *)row[i]));
break;
case TSDB_DATA_TYPE_SMALLINT:
len += sprintf(str + len, "%d", *((int16_t *)row[i]));
break;
case TSDB_DATA_TYPE_USMALLINT:
len += sprintf(str + len, "%u", *((uint16_t *)row[i]));
break;
case TSDB_DATA_TYPE_INT:
len += sprintf(str + len, "%d", *((int32_t *)row[i]));
break;
case TSDB_DATA_TYPE_UINT:
len += sprintf(str + len, "%u", *((uint32_t *)row[i]));
break;
case TSDB_DATA_TYPE_BIGINT:
len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_UBIGINT:
len += sprintf(str + len, "%" PRIu64, *((uint64_t *)row[i]));
break;
case TSDB_DATA_TYPE_FLOAT: {
float fv = 0;
fv = GET_FLOAT_VAL(row[i]);
len += sprintf(str + len, "%f", fv);
} break;
case TSDB_DATA_TYPE_DOUBLE: {
double dv = 0;
dv = GET_DOUBLE_VAL(row[i]);
len += sprintf(str + len, "%lf", dv);
} break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR: {
int32_t charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);
memcpy(str + len, row[i], charLen);
len += charLen;
} break;
case TSDB_DATA_TYPE_TIMESTAMP:
len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_BOOL:
len += sprintf(str + len, "%d", *((int8_t *)row[i]));
default:
break;
}
}
return len;
}
/**
* @brief print column name and values of each row
*
* @param res
* @return int
*/
static int printResult(TAOS_RES *res) {
int numFields = taos_num_fields(res);
TAOS_FIELD *fields = taos_fetch_fields(res);
char header[256] = {0};
int len = 0;
for (int i = 0; i < numFields; ++i) {
len += sprintf(header + len, "%s ", fields[i].name);
}
puts(header);
TAOS_ROW row = NULL;
while ((row = taos_fetch_row(res))) {
char temp[256] = {0};
printRow(temp, row, fields, numFields);
puts(temp);
}
}
int main() {
TAOS *taos = taos_connect("localhost", "root", "taosdata", "power", 6030);
if (taos == NULL) {
puts("failed to connect to server");
exit(EXIT_FAILURE);
}
TAOS_RES *res = taos_query(taos, "SELECT * FROM meters LIMIT 2");
if (taos_errno(res) != 0) {
printf("failed to execute taos_query. error: %s\n", taos_errstr(res));
exit(EXIT_FAILURE);
}
printResult(res);
taos_free_result(res);
taos_close(taos);
taos_cleanup();
}
// output:
// ts current voltage phase location groupid
// 1648432611249 10.300000 219 0.310000 California.SanFrancisco 2
// 1648432611749 12.600000 218 0.330000 California.SanFrancisco 2
<?php
use TDengine\Connection;
use TDengine\Exception\TDengineException;
try {
// instantiate
$host = 'localhost';
$port = 6030;
$username = 'root';
$password = 'taosdata';
$dbname = 'power';
$connection = new Connection($host, $port, $username, $password, $dbname);
// connect
$connection->connect();
$resource = $connection->query('SELECT ts, current FROM meters LIMIT 2');
var_dump($resource->fetch());
} catch (TDengineException $e) {
// throw exception
throw $e;
}
note
- 无论是使用 REST 连接还是原生连接的连接器,以上示例代码都能正常工作。
- 唯一需要注意的是:由于 REST 接口无状态, 不能使用
use db
语句来切换数据库。
异步查询
除同步查询 API 之外,TDengine 还提供性能更高的异步调用 API 处理数据插入、查询操作。在软硬件环境相同的情况下,异步 API 处理数据插入的速度比同步 API 快 2-4 倍。异步 API 采用非阻塞式的调用方式,在系统真正完成某个具体数据库操作前,立即返回。调用的线程可以去处理其他工作,从而可以提升整个应用的性能。异步 API 在网络延迟严重的情况下,优点尤为突出。
需要注意的是,只有使用原生连接的连接器,才能使用异步查询功能。
- Python
- C#
- C
import time
from ctypes import *
from taos import *
def fetch_callback(p_param, p_result, num_of_rows):
print("fetched ", num_of_rows, "rows")
p = cast(p_param, POINTER(Counter))
result = TaosResult(p_result)
if num_of_rows == 0:
print("fetching completed")
p.contents.done = True
result.close()
return
if num_of_rows < 0:
p.contents.done = True
result.check_error(num_of_rows)
result.close()
return None
for row in result.rows_iter(num_of_rows):
print(row)
p.contents.count += result.row_count
result.fetch_rows_a(fetch_callback, p_param)
def query_callback(p_param, p_result, code):
if p_result is None:
return
result = TaosResult(p_result)
if code == 0:
result.fetch_rows_a(fetch_callback, p_param)
result.check_error(code)
class Counter(Structure):
_fields_ = [("count", c_int), ("done", c_bool)]
def __str__(self):
return "{ count: %d, done: %s }" % (self.count, self.done)
def test_query(conn):
counter = Counter(count=0)
conn.query_a("select ts, current, voltage from power.meters", query_callback, byref(counter))
while not counter.done:
print(counter)
time.sleep(1)
print(counter)
conn.close()
if __name__ == "__main__":
test_query(connect())
# possible output:
# { count: 0, done: False }
# fetched 8 rows
# 1538548685000 10.300000 219
# 1538548695000 12.600000 218
# 1538548696800 12.300000 221
# 1538548696650 10.300000 218
# 1538548685500 11.800000 221
# 1538548696600 13.400000 223
# 1538548685500 10.800000 223
# 1538548686500 11.500000 221
# fetched 0 rows
# fetching completed
# { count: 8, done: True }
note
这个示例程序,目前在 Windows 系统上还无法运行
using System;
using System.Collections.Generic;
using TDengineDriver;
using TDengineDriver.Impl;
using System.Runtime.InteropServices;
namespace TDengineExample
{
public class AsyncQueryExample
{
static void Main()
{
IntPtr conn = GetConnection();
try
{
QueryAsyncCallback queryAsyncCallback = new QueryAsyncCallback(QueryCallback);
TDengine.QueryAsync(conn, "select * from meters", queryAsyncCallback, IntPtr.Zero);
Thread.Sleep(2000);
}
finally
{
TDengine.Close(conn);
}
}
static void QueryCallback(IntPtr param, IntPtr taosRes, int code)
{
if (code == 0 && taosRes != IntPtr.Zero)
{
FetchRawBlockAsyncCallback fetchRowAsyncCallback = new FetchRawBlockAsyncCallback(FetchRawBlockCallback);
TDengine.FetchRawBlockAsync(taosRes, fetchRowAsyncCallback, param);
}
else
{
throw new Exception($"async query data failed,code:{code},reason:{TDengine.Error(taosRes)}");
}
}
// Iteratively call this interface until "numOfRows" is no greater than 0.
static void FetchRawBlockCallback(IntPtr param, IntPtr taosRes, int numOfRows)
{
if (numOfRows > 0)
{
Console.WriteLine($"{numOfRows} rows async retrieved");
IntPtr pdata = TDengine.GetRawBlock(taosRes);
List<TDengineMeta> metaList = TDengine.FetchFields(taosRes);
List<object> dataList = LibTaos.ReadRawBlock(pdata, metaList, numOfRows);
for (int i = 0; i < dataList.Count; i++)
{
if (i != 0 && (i + 1) % metaList.Count == 0)
{
Console.WriteLine("{0}\t|", dataList[i]);
}
else
{
Console.Write("{0}\t|", dataList[i]);
}
}
Console.WriteLine("");
TDengine.FetchRawBlockAsync(taosRes, FetchRawBlockCallback, param);
}
else
{
if (numOfRows == 0)
{
Console.WriteLine("async retrieve complete.");
}
else
{
throw new Exception($"FetchRawBlockCallback callback error, error code {numOfRows}");
}
TDengine.FreeResult(taosRes);
}
}
static IntPtr GetConnection()
{
string host = "localhost";
short port = 6030;
string username = "root";
string password = "taosdata";
string dbname = "power";
var conn = TDengine.Connect(host, username, password, dbname, port);
if (conn == IntPtr.Zero)
{
throw new Exception("Connect to TDengine failed");
}
else
{
Console.WriteLine("Connect to TDengine success");
}
return conn;
}
}
}
// //output:
// // Connect to TDengine success
// // 8 rows async retrieved
// // 1538548685500 | 11.8 | 221 | 0.28 | california.losangeles | 2 |
// // 1538548696600 | 13.4 | 223 | 0.29 | california.losangeles | 2 |
// // 1538548685000 | 10.8 | 223 | 0.29 | california.losangeles | 3 |
// // 1538548686500 | 11.5 | 221 | 0.35 | california.losangeles | 3 |
// // 1538548685000 | 10.3 | 219 | 0.31 | california.sanfrancisco | 2 |
// // 1538548695000 | 12.6 | 218 | 0.33 | california.sanfrancisco | 2 |
// // 1538548696800 | 12.3 | 221 | 0.31 | california.sanfrancisco | 2 |
// // 1538548696650 | 10.3 | 218 | 0.25 | california.sanfrancisco | 3 |
// // async retrieve complete.
/**
* @brief call back function of taos_fetch_row_a
*
* @param param : the third parameter you passed to taos_fetch_row_a
* @param res : pointer of TAOS_RES
* @param numOfRow : number of rows fetched in this batch. will be 0 if there is no more data.
* @return void*
*/
void *fetch_row_callback(void *param, TAOS_RES *res, int numOfRow) {
printf("numOfRow = %d \n", numOfRow);
int numFields = taos_num_fields(res);
TAOS_FIELD *fields = taos_fetch_fields(res);
TAOS *_taos = (TAOS *)param;
if (numOfRow > 0) {
for (int i = 0; i < numOfRow; ++i) {
TAOS_ROW row = taos_fetch_row(res);
char temp[256] = {0};
printRow(temp, row, fields, numFields);
puts(temp);
}
taos_fetch_rows_a(res, fetch_row_callback, _taos);
} else {
printf("no more data, close the connection.\n");
taos_free_result(res);
taos_close(_taos);
taos_cleanup();
}
}
/**
* @brief callback function of taos_query_a
*
* @param param: the fourth parameter you passed to taos_query_a
* @param res : the result set
* @param code : status code
* @return void*
*/
void *select_callback(void *param, TAOS_RES *res, int code) {
printf("query callback ...\n");
TAOS *_taos = (TAOS *)param;
if (code == 0 && res) {
printHeader(res);
taos_fetch_rows_a(res, fetch_row_callback, _taos);
} else {
printf("failed to execute taos_query. error: %s\n", taos_errstr(res));
taos_free_result(res);
taos_close(_taos);
taos_cleanup();
exit(EXIT_FAILURE);
}
}
int main() {
TAOS *taos = taos_connect("localhost", "root", "taosdata", "power", 6030);
if (taos == NULL) {
puts("failed to connect to server");
exit(EXIT_FAILURE);
}
// param one is the connection returned by taos_connect.
// param two is the SQL to execute.
// param three is the callback function.
// param four can be any pointer. It will be passed to your callback function as the first parameter. we use taos
// here, because we want to close it after getting data.
taos_query_a(taos, "SELECT * FROM meters", select_callback, taos);
sleep(1);
}
// output:
// query callback ...
// ts current voltage phase location groupid
// numOfRow = 8
// 1538548685500 11.800000 221 0.280000 california.losangeles 2
// 1538548696600 13.400000 223 0.290000 california.losangeles 2
// 1538548685000 10.800000 223 0.290000 california.losangeles 3
// 1538548686500 11.500000 221 0.350000 california.losangeles 3
// 1538548685000 10.300000 219 0.310000 california.sanfrancisco 2
// 1538548695000 12.600000 218 0.330000 california.sanfrancisco 2
// 1538548696800 12.300000 221 0.310000 california.sanfrancisco 2
// 1538548696650 10.300000 218 0.250000 california.sanfrancisco 3
// numOfRow = 0
// no more data, close the connection.