前言:
最近在不定時的出差後,小編總結出來一個問題,自從2013年大數據火起來以後,到如今也經歷了6年多,有些公司已經逐步廢棄傳統的關係型數據庫,逐漸引入大數據平臺,有些公司可能已經經過大數據平臺從百億基本數據中找到的其真正的價值,可是這個過程老是緩慢的,總要經歷:數據的遷移、數倉的建設、基於數倉的上層開發、基於業務的報表分析...尤爲是在數據的遷移的過程當中,不一樣的項目根據需求不一樣會選擇適合本身的數據庫,從而致使,在將數據遷移到大數據平臺上時多種多樣的數據遷移方式,小編經歷過的就有MySQL、DB二、PostgreSQL、Oracle,像這些數據庫都是有直接的JDBC的驅動包,通常經過數據遷移工具sqoop或者一些ETL工具直接能夠將數據抽取過來,可是也有些並不是有JDBC的驅動包,有些數據遷移工具也並不支持的數據庫;好比接下來小編要介紹的InfluxDB,他就是一個時序數據庫,並且並無相應比較好用的工具去抽取數據。只能先了解其原理,而後本身想辦法,如何高效、簡單的實現數據的遷移。
數據遷移老是一個頭疼的問題,這裏小編建議,先根據本身的業務和需求,在遷移數據以前,設計出適合各類場景的表,千萬不要盲目的把數據遷移過來,以後再改表類型以及結構,不然後期的維護成本就至關高。好了,廢話很少說了,以上是小編的一點愚見,接下來開始學習這個InfluxDB時序數據庫!html
InfluxDB 是用Go語言編寫的一個開源分佈式時序、事件和指標數據庫,無需外部依賴。 與 Elasticsearch 有些相似。
功能:
- 基於時間序列,支持與時間有關的相關函數(如最大,最小,求和等);
- 可度量性:你能夠實時對大量數據進行計算;
- 基於事件:它支持任意的事件數據基於事件:它支持任意的事件數據。
主要特色:
- 無結構(無模式):能夠是任意數量的列
- 可拓展的,支持min, max, sum, count, mean, median 等一系列函數,方便統計支持min, max, sum, count, mean, median 等一系列函數,方便統計
- 原生的HTTP支持,內置HTTP API原生的HTTP支持,內置HTTP API
- 強大的類SQL語法強大的類SQL語法
- 自帶管理界面,方便使用自帶管理界面,方便使用
InfluxDB與傳統數據庫的比較:
java
接下來經過一個insert操做,展開對InfluxDB獨有概念的介紹,在 InfluxDB 中,咱們能夠粗略的將要存入的一條數據看做一個虛擬的 key 和其對應的 value(field value),格式以下:redis
insert cpu_usage,host=server01,region=us-west value=0.64 1434055562000000000`
虛擬的 key 包括如下幾個部分: database, retention policy, measurement, tag sets, field name, timestamp。數據庫
Point 由時間戳(time)、數據(field)、標籤(tags)組成。 vim
Point至關於傳統數據庫裏的一行數據 , 以下表所示:
緩存
Series 至關因而 InfluxDB 中一些數據的集合,在同一個 database 中,retention policy、measurement、tag sets 徹底相同的數據同屬於一個 series,同一個 series 的數據在物理上會按照時間順序排列存儲在一塊兒。服務器
Shard 在 InfluxDB 中是一個比較重要的概念,它和 retention policy 相關聯。每個存儲策略下會存在許多 shard,每個 shard 存儲一個指定時間段內的數據,而且不重複 ; 例如 : 7點-8點 的數據落入 shard0 中,8點-9點的數據則落入 shard1 中。每個 shard 都對應一個底層的 tsm 存儲引擎,有獨立的 cache、wal、tsm file。併發
TSM 存儲引擎主要由幾個部分組成:cache、wal、tsm file、compactor 。
1)Cache :cache 至關因而 LSM Tree 中的 memtabl。插入數據時,其實是同時往 cache 與 wal 中寫入數據,能夠認爲 cache 是 wal 文件中的數據在內存中的緩存。當 InfluxDB 啓動時,會遍歷全部的 wal 文件,從新構造 cache,這樣即便系統出現故障,也不會致使數據的丟失。
cache 中的數據並非無限增加的,有一個 maxSize 參數用於控制當 cache 中的數據佔用多少內存後就會將數據寫入 tsm 文件。若是不配置的話,默認上限爲 25MB,每當 cache 中的數據達到閥值後,會將當前的 cache 進行一次快照,以後清空當前 cache 中的內容,再建立一個新的 wal 文件用於寫入,剩下的 wal 文件最後會被刪除,快照中的數據會通過排序寫入一個新的 tsm 文件中 。
2)WAL:wal 文件的內容與內存中的 cache 相同,其做用就是爲了持久化數據,當系統崩潰後能夠經過 wal 文件恢復尚未寫入到 tsm 文件中的數據 。
3) TSM File : 單個 tsm file 大小最大爲 2GB,用於存放數據 。
4) Compactor:compactor 組件在後臺持續運行,每隔 1 秒會檢查一次是否有須要壓縮合並的數據 。
主要進行兩種操做 :
- 一種是 cache 中的數據大小達到閥值後,進行快照,以後轉存到一個新的 tsm 文件中 。
- 另一種就是合併當前的 tsm 文件,將多個小的 tsm 文件合併成一個,使每個文件儘可能達到單個文件的最大大小,減小文件的數量,而且一些數據的刪除操做也是在這個時候完成 。
app
官網地址:https://dl.influxdata.comcurl
#1.本地下載 wget https://dl.influxdata.com/influxdb/releases/influxdb-1.1.0.x86_64.rpm yum localinstall influxdb-1.1.0.x86_64.rpm #2.在線yum 安裝 #2.1配置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 #2.2 安裝 yum install go sudo yum install influxdb
#啓動服務 systemctl start influxdb.service #查看服務是否正常 systemctl status influxdb #查看服務對應進程 ps aux | grep influx
Ps:
(1)8086端口:HTTP API的端口
(2)8088:備份和恢復時使用,默認是8088
因爲是使用yum 安裝的因此安裝後,Influx的目錄會分佈在 /usr/bin 、 /var/lib/influxdb/ 、 /etc/influxdb/ 下,下面咱們一一介紹:
(1)/usr/bin 該目錄下是存放相應命令操做的目錄:
influxd influxdb服務器 influx influxdb命令行客戶端 influx_inspect 查看工具 influx_stress 壓力測試工具 influx_tsm 數據庫轉換工具(將數據庫從b1或bz1格式轉換爲tsm1格式)
(2)/var/lib/influxdb/ 存放數據的目錄
data 存放最終存儲的數據,文件以.tsm結尾 meta 存放數據庫元數據 wal 存放預寫日誌文件
(3)/etc/influxdb/influxdb.conf 存放配置文件的目錄
influxdb.conf 就是 influxdb的配置文件。
Ps:在閱讀配置參數的時,最好是對該數據庫各個概念以及原理有一些瞭解,再去細看配置參數。
#編輯配置文件 vim /etc/influxdb/influxdb.conf #如下=後面的都是默認值 reporting-disabled = false -- 該選項用於上報influxdb的使用信息給InfluxData公司 bind-address = "127.0.0.1:8088" -- 備份恢復時使用,默認值爲8088 [meta]下 dir = "/var/lib/influxdb/meta" -- meta數據存放目錄 retention-autocreate = true -- 用於控制默認存儲策略 logging-enabled = true -- 是否開啓meta日誌 [data]下 dir = "/var/lib/influxdb/data" -- 最終數據(TSM文件)存儲目錄 wal-dir = "/var/lib/influxdb/wal" -- 預寫日誌存儲目錄 query-log-enabled = true -- 是否開啓tsm引擎查詢日誌 cache-max-memory-size = "1g" -- 用於限定shard最大值,大於該值時會拒絕寫入 cache-snapshot-memory-size = "25m" -- 用於設置快照大小,大於該值時數據會刷新到tsm文件 cache-snapshot-write-cold-duration = "10m" -- tsm1引擎 snapshot寫盤延遲 compact-full-write-cold-duration = "4h" -- tsm文件在壓縮前能夠存儲的最大時間 max-series-per-database = 1000000 -- 限制數據庫的級數,該值爲0時取消限制 trace-logging-enabled = false -- 是否開啓trace日誌 [coordinator] 下 write-timeout = "10s" -- 寫操做超時時間 max-concurrent-queries = 0 -- 最大併發查詢數,0無限制 query-timeout = "0s" -- 查詢操做超時時間,0無限制 log-queries-after = "0s" -- 慢查詢超時時間,0無限制 max-select-point = 0 -- SELECT語句能夠處理的最大點數(points)0無限制 max-select-series = 0 -- SELECT語句能夠處理的最大級數(series),0無限制 max-select-buckets = 0 -- SELECT語句能夠處理的最大"GROUP BY time()"的時間週期,0無限制 [retention]下 ,舊數據的保留策略 enabled = true -- 是否開啓該模塊 check-interval = "30m" -- 檢查時間間隔 [http] 下,influxdb的http接口配置 enabled = true -- 是否開啓該模塊 bind-address = ":8086" --綁定地址 auth-enabled = false -- 是否開啓認證 log-enabled = true -- 是否開啓日誌 max-row-limit = 0 -- 配置查詢返回最大行數 max-connection-limit = 0 -- 配置最大鏈接數,0無限制
以上是常見操做配置,具體細節請參考:
https://www.cnblogs.com/MikeZhang/p/InfluxDBInstall20170206.html
Ps:如下操做可能與關係型數據庫操做不一樣,若是對Influx不瞭解,請先閱讀Influx介紹以後,在繼續往下閱讀。
#進入Influx數據庫 [root@iZbp19ujl2isnn8zc1hqirZ ~]# influx > show databases #顯示全部數據庫 > create database tes #建立數據庫 > drop database test #刪除數據庫 > use test #進入數據庫 > insert disk_free,hostname=server01 value=442221834240i #建立&& 插入數據 > select * from disk_free #查詢數據 > show measurement #顯示庫中的全部表 > drop measurement disk_free #刪除表
咱們發現以上猛如虎的操做中,進入沒有相似create table的命令,這是爲何呢?
原來是由於: InfluxDB中沒有顯示的建立表的語句,只能經過insert數據的房仍是來創建新表 。
insert disk_free,hostname=server01 value=442221834240i -- 剖析以上命令的含義 disk_free 就是表名,hostname 是索引(tag),value=xx 是記錄值(field),記錄值能夠有多個,系統自帶追加時間戳。 -- 也能夠手動添加時間戳 insert disk_free,hostname=server01 value=442221834240i 1435362189575692182
介紹: InfluxDB 是沒有提供直接刪除數據記錄的方法,可是提供數據保存策略,主要用於指定數據保留時間,超過指定時間,就刪除這部分數據。
#查看當前數據庫中的Retention Policies >show retention policies on test name duration shardGroupDuration replicaN default ---- -------- ------------------ -------- ------- autogen 0s 168h0m0s 1 false
解釋:
- name:名稱,此示例名稱爲 default。
- duration:持續時間,0表明無限制。
- shardGroupDuration:shardGroup的存儲時間,shardGroup是InfluxDB的一個基本儲存結構,應該大於這個時間的數據在查詢效率上應該有所下降。
- replicaN:全稱是replication,副本個數。
- default:是不是默認策略。
#建立新的Retention Policies > create retention policy "rp_name" on "test" duration 3w replication 1 default #修改Retention Policies > alter retention policy "rp_name" on "test" duration 30d default > show retention policies on test name duration shardGroupDuration replicaN default ---- -------- ------------------ -------- ------- autogen 0s 168h0m0s 1 false rp_name 720h0m0s 24h0m0s 1 true #刪除Retention Policies drop retention policy "rp_name" on "test"
建立語句剖析:
create retention policy "rp_name" on "test" duration 3w replication 1 default
- rp_name:保存策略名稱
- test:所針對的數據庫
- 3w : 保存3周,3周以前的數據將被刪除,influxdb 具有各類事件參數,持續時間必須至少爲1小時;好比:h(小時)、d(天)、w(星期) 。
- replication : 副本個數,通常爲1便可。
介紹: InfluxDB 的連續查詢是在數據庫中自動定時啓動的一組語句,語句中必須包含 select 關鍵字 和 group by time() 關鍵字。 InfluxDB 會將查詢結果放在指定的數據表中。
目的:使用連續查詢是最優的下降採樣率的方式,連續查詢和存儲策略搭配使用將會大大下降 InfluxDB 的系統佔用量。並且使用連續查詢後,數據會存放到指定的數據表中,這樣就爲之後統計不一樣精度的數據提供了方便。
建立語句:
CREATE CONTINUOUS QUERY <cq_name> ON <database_name> [RESAMPLE [EVERY <interval>] [FOR <interval>]] BEGIN SELECT <function>(<stuff>)[,<function>(<stuff>)] INTO <different_measurement> FROM <current_measurement> [WHERE <stuff>] GROUP BY time(<interval>)[,<stuff>] END
舉例:
CREATE CONTINUOUS QUERY wj_30m ON test BEGIN SELECT mean(connected_clients), MEDIAN(connected_clients), MAX(connected_clients), MIN(connected_clients) INTO redis_clients_30m FROM redis_clients GROUP BY ip,port,time(30m) --解釋: 在test數據庫中新建了一個名爲 wj_30m 的連續查詢,每三十分鐘取一個 connected_clients 字段的平均值、中位值、最大值、最小值從redis_clients表中而且插入到redis_clients_30m表中,使用的數據保留策略都是default。
連續查詢的其餘操做:
#查看庫中的連續查詢 > show continuous queries name: _internal name query ---- ----- name: test name query ---- ----- #刪除Continuous Queries > drop continuous query <cq_name> on <database_name>
用戶管理:
#以xxx用戶登陸 $influx -username useer -password abcd #顯示全部用戶 > show users user admin ---- ----- zy true #建立普通用戶 > CREATE USER "username" WITH PASSWORD 'password' #建立管理員用戶 > CREATE USER "admin" WITH PASSWORD 'admin' WITH ALL PRIVILEGES #爲用戶設置密碼 > SET PASSWORD FOR <username> = '<password>' #刪除用戶 > DROP USER "username"
權限設置:
#爲一個已有用戶受權管理員權限 > GRANT ALL PRIVILEGES TO <username> #取消用戶權限 > REVOKE ALL PRIVILEGES FROM <username> #展現用戶在不一樣數據庫上的權限 > SHOW GRANTS FOR <user_name>
關於Influxdb支持兩種方式:類SQL查詢和Http接口查詢:
-- 類SQL查詢(詢最新的三條數據) SELECT * FROM weather ORDER BY time DESC LIMIT 3 #Http接口查詢 $curl -G 'http://localhost:8086/query?pretty=true' --data-urlencode "db=test" --data-urlencode "q=SELECT * FROM weather ORDER BY time DESC LIMIT 3"
這裏小編以maven項目的結構,測試關於InfluxDB數據庫的增刪改查。
<!-- InfluxDB 須要的jar包 --> <dependency> <groupId>org.influxdb</groupId> <artifactId>influxdb-java</artifactId> <version>2.5</version> </dependency>
InfluxDBUtils:
import org.influxdb.InfluxDB; import org.influxdb.InfluxDBFactory; import org.influxdb.dto.Point; import org.influxdb.dto.Query; import org.influxdb.dto.QueryResult; import java.util.Map; /** * * Created with IntelliJ IDEA. * * User: ZZY * * Date: 2019/11/15 * * Time: 10:10 * * Description: */ public class InfluxDBConnect { private String username;//用戶名 private String password;//密碼 private String openurl;//鏈接地址 private String database;//數據庫 private InfluxDB influxDB; public InfluxDBConnect(String username, String password, String openurl, String database){ this.username = username; this.password = password; this.openurl = openurl; this.database = database; } /**鏈接時序數據庫;得到InfluxDB**/ public InfluxDB getConnect(){ if(influxDB==null){ influxDB=InfluxDBFactory.connect(openurl,username,password); influxDB.createDatabase(database); } return influxDB; } /** * 設置數據保存策略 * defalut 策略名 /database 數據庫名/ 30d 數據保存時限30天/ 1 副本個數爲1/ 結尾DEFAULT 表示 設爲默認的策略 */ public void setRetentionPolicy(){ String command=String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT", "defalut", database, "30d", 1); this.query(command); } /** * 查詢 * @param command 查詢語句 * @return */ public QueryResult query(String command){ return influxDB.query(new Query(command,database)); } /** * 插入 * @param measurement 表 * @param tags 標籤 * @param fields 字段 */ public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields){ Point.Builder builder =Point.measurement(measurement); builder.tag(tags); builder.fields(fields); influxDB.write(database,"",builder.build()); } /** * 刪除 * @param command 刪除語句 * @return 返回錯誤信息 */ public String deleteMeasurementData(String command){ QueryResult query = influxDB.query(new Query(command, database)); return query.getError(); } /** * 建立數據庫 * @param dbName */ public void createDB(String dbName){ influxDB.createDatabase(dbName); } /** * 刪除數據庫 * @param dbName */ public void deleteDB(String dbName){ influxDB.deleteDatabase(dbName); } }
pojo:
import java.io.Serializable; /** * * Created with IntelliJ IDEA. * * User: ZZY * * Date: 2019/11/15 * * Time: 10:07 * * Description: */ public class CodeInfo implements Serializable { private static final long serialVersionUID = 1L; private Long id; private String name; private String code; private String descr; private String descrE; private String createdBy; private Long createdAt; private String time; private String tagCode; private String tagName; public static long getSerialVersionUID() { return serialVersionUID; } } //set and get method ...
測試:
import org.influxdb.InfluxDB; import org.influxdb.dto.QueryResult; import java.util.*; /** * * Created with IntelliJ IDEA. * * User: ZZY * * Date: 2019/11/15 * * Time: 11:45 * * Description: 測試influxDB的增刪改查 */ public class Client { public static void main(String[] args) { String username = "admin";//用戶名 String password = "admin";//密碼 String openurl = "http://192.168.254.100:8086";//鏈接地址 String database = "test";//數據庫 InfluxDBConnect influxDBConnect = new InfluxDBConnect(username, password, openurl, database); influxDBConnect.getConnect(); //insertInfluxDB(influxDBConnect); testQuery(influxDBConnect); } //向Measurement中插入數據 public static void insertInfluxDB(InfluxDBConnect influxDB) { Map<String, String> tags = new HashMap<String, String>(); Map<String, Object> fields = new HashMap<String, Object>(); List<CodeInfo> list = new ArrayList<CodeInfo>(); CodeInfo info1 = new CodeInfo(); info1.setId(1L); info1.setName("BANKS"); info1.setCode("ABC"); info1.setDescr("中國農業銀行"); info1.setDescrE("ABC"); info1.setCreatedBy("system"); info1.setCreatedAt(new Date().getTime()); CodeInfo info2 = new CodeInfo(); info2.setId(2L); info2.setName("BANKS"); info2.setCode("CCB"); info2.setDescr("中國建設銀行"); info2.setDescrE("CCB"); info2.setCreatedBy("system"); info2.setCreatedAt(new Date().getTime()); list.add(info1); list.add(info2); String measurement = "sys_code"; for (CodeInfo info : list) { tags.put("TAG_CODE", info.getCode()); tags.put("TAG_NAME", info.getName()); fields.put("ID", info.getId()); fields.put("NAME", info.getName()); fields.put("CODE", info.getCode()); fields.put("DESCR", info.getDescr()); fields.put("DESCR_E", info.getDescrE()); fields.put("CREATED_BY", info.getCreatedBy()); fields.put("CREATED_AT", info.getCreatedAt()); influxDB.insert(measurement, tags, fields); } } //查詢Measurement中的數據 public static void testQuery(InfluxDBConnect influxDB) { String command = "select * from sys_code"; QueryResult results = influxDB.query(command); if (results == null) { return; } for(QueryResult.Result result:results.getResults()){ List<QueryResult.Series> series = result.getSeries(); for(QueryResult.Series serie :series){ System.out.println("serie:"+serie.getName()); //表名 Map<String, String> tags =serie.getTags(); if(tags !=null){ System.out.println("tags:-------------------------"); tags.forEach((key, value)->{ System.out.println(key + ":" + value); }); } System.out.println("values:-----------------------"); List<List<Object>> values = serie.getValues(); //列出每一個serie中全部的列--value 列爲全大寫 List<String> columns =serie.getColumns(); //列出每一個serie中全部的列 for(List<Object> list : values){ for(int i=0; i< list.size(); i++){ String propertyName = setColumns(columns.get(i));//字段名 Object value =list.get(i); System.out.println(value.toString()); } } System.out.println("columns:"); for(String column:columns){ System.out.println(column); } } } } //刪除Measurement中的數據 public static void deletMeasurementData(InfluxDBConnect influxDB){ String command = "delete from sys_code where TAG_CODE='ABC'"; String err =influxDB.deleteMeasurementData(command); System.out.println(err); } private static String setColumns(String column){ System.out.println(column); String[] cols = column.split("_"); StringBuffer sb = new StringBuffer(); for(int i=0; i< cols.length; i++){ String col = cols[i].toLowerCase(); if(i != 0){ String start = col.substring(0, 1).toUpperCase(); String end = col.substring(1).toLowerCase(); col = start + end; } sb.append(col); } System.out.println(sb.toString()); return sb.toString(); } }
(1)普通導出
$influx_inspect export -datadir "/var/lib/influxdb/data" -waldir "/var/lib/influxdb/wal" -out "test_sys" -database "test" -start 2019-07-21T08:00:01Z #命令解釋 influx_inspect export -datadir "/data/influxdb/data" # 勿動,influxdb 默認的數據存儲位置 -waldir "/data/influxdb/wal" # 勿動,influxdb 默認的數據交換位置 -out "telemetry_vcdu_time" # 導出數據文件的文件名 -database telemetry_vcdu_time # 指定要導出數據的數據庫 -start 2019-07-21T08:00:01Z # 指定要導出的數據的起始時間
此時在當前目錄下會出現一個名爲test_sys的文件,查看文件內容:
(2)導出成CSV格式文件
$influx -database 'test' -execute 'select * from sys_code' -format='csv' > sys_code.csv
此時在當前目錄下就多出一個sys_code.csv的文件,查看文件內容:
$influx -import -path=telemetry_sat_time -precision=ns #命令解釋 influx -import # 無參,勿動 -path=telemetry_sat_time # 指定導入數據的文件 -precision=ns # 指定導入數據的時間精度