# Storm OpenTSDB 集成 ## Storm OpenTSDB Bolt 和 TridentState OpenTSDB 為時間序列數據提供了可擴展且高可用性的存儲. 它由 Time Series Daemon (TSD) servers 以及命令行工具組成. 每個 TSD 連接到配置的 HBase 集群以 push/query（推送/查詢） 數據. 時間序列數據點包括: - a metric name. - a UNIX timestamp (seconds or milliseconds since Epoch). - a value (64 bit integer or single-precision floating point value). - a set of tags (key-value pairs) that describe the time series the point belongs to. Storm bolt 和 trident state 從一個基于給定的 `TupleMetricPointMapper` 的 tuple 中創建了上面的時間序列數據. 該模塊提供了 core Storm 和 Trident bolt 實現, 用戶將數據寫入 OpenTSDB. 時間序列數據點被寫入時有 at-least-once（至少一次）的語義保證, 并且重復的數據點應該像 OpenTSDB 的 [這里](http://opentsdb.net/docs/build/html/user_guide/writing.html#duplicate-data-points) 提及的一樣來處理. ## 示例 ### Core Bolt 下面的示例描述了 `org.apache.storm.opentsdb.bolt.OpenTsdbBolt` cord bolt 的使用方法 ``` OpenTsdbClient.Builder builder = OpenTsdbClient.newBuilder(openTsdbUrl).sync(30_000).returnDetails(); final OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(builder, Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); openTsdbBolt.withBatchSize(10).withFlushInterval(2000); topologyBuilder.setBolt("opentsdb", openTsdbBolt).shuffleGrouping("metric-gen"); ``` ### Trident State ``` final OpenTsdbStateFactory openTsdbStateFactory = new OpenTsdbStateFactory(OpenTsdbClient.newBuilder(tsdbUrl), Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); TridentTopology tridentTopology = new TridentTopology(); final Stream stream = tridentTopology.newStream("metric-tsdb-stream", new MetricGenSpout()); stream.peek(new Consumer() { @Override public void accept(TridentTuple input) { LOG.info("########### Received tuple: [{}]", input); } }).partitionPersist(openTsdbStateFactory, MetricGenSpout.DEFAULT_METRIC_FIELDS, new OpenTsdbStateUpdater()); ```