MDD

度量驱动开发 Metrics-Driven Development 即在软件开发过程中, 设计和实现方案的选择及优化建立在度量的基础之上.
一切拿数据说话, 杜绝臆想和猜测, 这是在大数据和 DevOPS 浪潮之下顺应而生的现代软件开发方法

• 监控: 采集产品的性能, 用量以及业务相关的统计数据
• 存储: 把收集到的数据基于时间线或其他基准分门别类进行存储
• 分析: 对于数据进行各种维度的分析计算, 生成图表, 并对关键指标设置阀值, 在条件满足时触发报警

牵涉到的开源工具不胜枚举, 常用的就有

• LogStash
• Kibana
• ElasticSearch
• Collectd
• Metrics
• Graphite
• InfluxDB
• Grafana
• Splunk
• Circonus
• Seyren
• New Relic
• Cassandra

我们先从时间序列数据库 InfluxDB 说起

时间序列数据

光阴似箭, 日月如梭, 很多事情都可以重头再来, 唯有时间一去不回头.

在软件应用领域, 一寸光阴一寸金, 基于时间的数据收集, 度量和监控每时每刻都在进行.

比如股指的变化, 房价的涨跌, 天气的冷暖, 应用访问的频率和次数等等
这些数据都有一个共通的地方, 那就是全部都是以时间为基轴

我们称这些数据为 Time-series Data 时间序列数据, 由一系列来自相同数据源, 按一定时间间隔采样的连续数据组成.

这些数据的重要性不言自明, 在 DevOps 的大潮下, 人们逐渐达成共识, 一定要基于数据说话和做决定

时间序列数据的存储可以是数据库, 主键是时间点, 字段是具体的数据值.
但有由于时间序列数据有它自身的特点, 就如 Baron Schwartz 所总结的

1. 90%以上的数据库负载在于大量和高频率的写操作
2. 写操作一般都是在以有的数据度量之后按时间追加
3. 这些写操作都是典型地按时间序列产生的, 比如每秒或每分
4. 数据库的主要瓶颈在于输入输出的限制
5. 对于已存在的单个数据点的纠正和修改极少
6. 删除数据几乎都是一个比较大的时间段(天, 月或年), 极少会只针对单条记录
7. 查询数据库一般都是基于某个序列的连续数据, 并根据时间或时间的函数排序
8. 执行查询常用并发读取一个或多个序列

对于时间序列数据当然可以使用传统的数据库来存储, 比如Oracle, MySQL, PostgreSQL, 或者一些KV store, 比如 Cassandra, Riak 等

但是对于上述时间序列数据的特点有针对性的优化的一些时间序列数据库逐渐流行起来, 比如

• InfluxDB
• OpenTSDB (Runs on top of Hadoop and HBase)
• KDB+
• Graphite

时间序列数据库 InfluxDB

这里就重点讲讲现在比较流行的 InfluxDB , 它是由InfluxData开发的一个开源的时间序列数据库。 它是由 Go 语言写的， 为了快速并且高可靠性的时间序列数据存取做了一些优化， 应用范围包括操作监视， 应用度量， 物联传感器数据以及实时分析等方面。

概述

首先 InfluxDB是一个基于时间序列数据组织的，例如 cpu占用率, 温度等度量数据， 在时间序列上有一个或多个数据在连续的时间间隔上的采样

理论上讲， 你可以认为 measurement 就象传统数据库中的表 table, 它的主键永远是 time, tags 和 fields 是表中的字段， tags 是有索引的字段，而 fields 没有.

横向比较一下 InfluxDBI, Oracle 和 Cassandra

InfluxDB	Oracle	Cassandra
Measurement	Table	Column Family
Timestamp as PK	Customized PK	Parition Key
Tag	Indexed field	Clustering column
Field	not Indexed field	column

区别在于， InfluxDB 中可以有数百万的 measurements, 无需事先定义 schema, 不会有 null 类型的数据， 数据点 Point 写到 InfluxDB 中是采用 Line 协议， 即如下格式

<measurement>[,<tag-key>=<tag-value>...] <field-key>=<field-value>[,<field2-key>=<field2-value>...] [unix-nano-timestamp]

下面就是一个例子

cpu,host=serverA,region=us_west value=0.64payment,device=mobile,product=Notepad,method=credit billed=33,licenses=3i 1434067467100293230stock,symbol=AAPL bid=127.46,ask=127.48temperature,machine=unit42,type=assembly external=25,internal=37 1434067467000000000

** 注:** 更多有关 line 协议的信息可参见 行协议书写语法.

存储的数据结构大体如下

{
"database" : "mydb", 
"retentionPolicy" : "bar", 
"points" : 
    [ 
        {"name" : "disk", 
        "tags" : {"server" : "bwi23", "unit" : "1"}, 
        "timestamp" : "2015-03-16T01:02:26.234Z", 
        "fields" : {"total" : 100, "used" : 40, "free" : 60}}
    ]
}

安装

MAC OS

brew install influxdb

CentOS

先添加一个 yum 配置文件

cat <<EOF | sudo tee /etc/yum.repos.d/influxdb.repo
[influxdb]
name = InfluxDB Repository - RHEL \$releasever
baseurl = https://repos.influxdata.com/rhel/\$releasever/\$basearch/stable
enabled = 1
gpgcheck = 1
gpgkey = https://repos.influxdata.com/influxdb.key
EOF

测试

打开 http://localhost:8083/，这是它的 Web 控制台
(注: /etc/influxdb/influxdb.conf 中的# https-enabled = false 要改成 true)

还有两种方式访问influxdb, 命令行方式和API方式

• http://localhost:8086/
• influx 命令行工具

用法

创建数据库

curl -i -XPOST http://localhost:8086/query --data-urlencode "q=CREATE DATABASE mydb"

或者

influx -execute 'create database mydb'

查询数据结构

• 查看 databases

SHOW DATABASES

• 查看 retention policies

 SHOW RETENTION POLICIES

• 查看 series

 SHOW SERIES

• 查看 measurements

SHOW MEASUREMENTS

• 查看 tag keys

SHOW TAG KEYS

• 查看 tag values

SHOW TAG VALUES

• 查看 field keys

SHOW FIELD KEYS

插入数据

curl -XPOST 'http://localhost:8086/write?db=mydb' \\\\\\\\\\\\\\\\-d 'cpu,host=server01,region=uswest load=42 1434055562000000000'curl -XPOST 'http://localhost:8086/write?db=mydb' \\\\\\\\\\\\\\\\-d 'cpu,host=server02,region=uswest load=78 1434055562000000000'curl -XPOST 'http://localhost:8086/write?db=mydb' \\\\\\\\\\\\\\\\-d 'cpu,host=server03,region=useast load=15.4 1434055562000000000'

或

influx -execute 'INSERT into mydb.autogen cpu,host=serverA,region=us_west value=0.64'

查询数据

influx -execute 'show databases'
influx -execute 'show measurements' -database mydb
influx -execute 'SELECT host, region, value FROM cpu' -database mydb

或

curl -G 'http://localhost:8086/query?db=mydb' --data-urlencode "q=SHOW MEASUREMENTS"

curl -G http://localhost:8086/query?pretty=true --data-urlencode "db=mydb" --data-urlencode "q=SELECT * FROM cpu WHERE host='server01' AND time < now() - 1d"

分析数据

curl -G http://localhost:8086/query?pretty=true --data-urlencode "db=mydb" --data-urlencode "q=SELECT mean(load) FROM cpu WHERE region='uswest'"

InfluxDB Client library

Java Client library

Maven 依赖

<dependency>
  <groupId>org.influxdb</groupId>
  <artifactId>influxdb-java</artifactId>
  <version>2.4</version>
</dependency>

示例

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

import java.util.concurrent.TimeUnit;

public class InfluxDbSample {


    public static void main(String[] args) {
        InfluxDB influxDB = InfluxDBFactory.connect("http://127.0.0.1:8086", "root", "");
        Pong pong = influxDB.ping();
        System.out.println("pong: " + pong);

        String dbName = "aTimeSeries";
        influxDB.createDatabase(dbName);

        // Flush every 2000 Points, at least every 100ms
        //influxDB.enableBatch(2000, 100, TimeUnit.MILLISECONDS);
        System.out.println("--- created database and will insert data ---");

        BatchPoints batchPoints = BatchPoints
                .database(dbName)
                .tag("async", "true")
                .retentionPolicy("default")
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        Point point1 = Point.measurement("cpu")
                .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                .addField("idle", 90L)
                .addField("user", 9L)
                .addField("system", 1L)
                .build();
        Point point2 = Point.measurement("disk")
                .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                .addField("used", 80L)
                .addField("free", 1L)
                .build();
        batchPoints.point(point1);
        batchPoints.point(point2);
        influxDB.write(batchPoints);
        Query query = new Query("SELECT idle FROM cpu", dbName);
        QueryResult rs = influxDB.query(query);

        System.out.println("result: " + rs.toString());
        influxDB.deleteDatabase(dbName);
    }
}

--- output ---

pong: Pong{version=0.13.0, responseTime=245}
--- created database and will insert data ---
result: QueryResult [results=[Result [series=[Series [name=cpu, tags=null, columns=[time, idle], values=[[2016-11-03T14:51:42.486Z, 90.0], [2016-11-03T14:52:33.328Z, 90.0], [2016-11-03T14:54:03.474Z, 90.0], [2016-11-03T14:55:55.839Z, 90.0], [2016-11-03T15:00:15.402Z, 90.0], [2016-11-03T15:02:50.596Z, 90.0]]]], error=null]], error=null]

• 参见 https://github.com/influxdata/influxdb-java

实用链接

• Google group of InfluxDB
• InfluxDB 相关术语

参考文章

• What is time series data
• Basic Time Series with Cassandra
• Advanced Time Series with Cassandra
• Cassandra Time Series Bucketing
• Basic rules of cassandra data modeling