You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@griffin.apache.org by gu...@apache.org on 2018/06/14 09:01:56 UTC
incubator-griffin git commit: refactor persist to sink
Repository: incubator-griffin
Updated Branches:
refs/heads/master 88ac8f7cd -> 23a3e999e
refactor persist to sink
Author: William Guo <gu...@apache.org>
Closes #302 from guoyuepeng/2018_06_14_refactor_persist.
Project: http://git-wip-us.apache.org/repos/asf/incubator-griffin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-griffin/commit/23a3e999
Tree: http://git-wip-us.apache.org/repos/asf/incubator-griffin/tree/23a3e999
Diff: http://git-wip-us.apache.org/repos/asf/incubator-griffin/diff/23a3e999
Branch: refs/heads/master
Commit: 23a3e999eb522850b23d41fb0ea7516d133fc921
Parents: 88ac8f7
Author: William Guo <gu...@apache.org>
Authored: Thu Jun 14 17:01:41 2018 +0800
Committer: Lionel Liu <bh...@163.com>
Committed: Thu Jun 14 17:01:41 2018 +0800
----------------------------------------------------------------------
.../griffin/measure/context/DQContext.scala | 12 +-
.../measure/context/sink/ConsoleSink.scala | 84 +++++++++
.../griffin/measure/context/sink/HdfsSink.scala | 186 +++++++++++++++++++
.../griffin/measure/context/sink/HttpSink.scala | 82 ++++++++
.../measure/context/sink/MongoSink.scala | 128 +++++++++++++
.../measure/context/sink/MultiSinks.scala | 85 +++++++++
.../griffin/measure/context/sink/Sink.scala | 47 +++++
.../measure/context/sink/SinkFactory.scala | 58 ++++++
.../measure/context/sink/SinkTaskRunner.scala | 101 ++++++++++
.../measure/context/writer/HdfsPersist.scala | 186 -------------------
.../measure/context/writer/HttpPersist.scala | 82 --------
.../measure/context/writer/LoggerPersist.scala | 84 ---------
.../measure/context/writer/MongoPersist.scala | 128 -------------
.../measure/context/writer/MultiPersists.scala | 85 ---------
.../measure/context/writer/Persist.scala | 47 -----
.../measure/context/writer/PersistFactory.scala | 58 ------
.../context/writer/PersistTaskRunner.scala | 101 ----------
17 files changed, 777 insertions(+), 777 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/DQContext.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/DQContext.scala b/measure/src/main/scala/org/apache/griffin/measure/context/DQContext.scala
index 926bde1..5b9cda6 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/context/DQContext.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/DQContext.scala
@@ -21,7 +21,7 @@ package org.apache.griffin.measure.context
import org.apache.griffin.measure.configuration.enums._
import org.apache.griffin.measure.configuration.params._
import org.apache.griffin.measure.datasource._
-import org.apache.griffin.measure.context.writer._
+import org.apache.griffin.measure.context.sink._
import org.apache.spark.sql.{Encoders, SQLContext, SparkSession}
/**
@@ -63,14 +63,14 @@ case class DQContext(contextId: ContextId,
if (dataSourceNames.size > index) dataSourceNames(index) else ""
}
- private val persistFactory = PersistFactory(persistParams, name)
- private val defaultPersist: Persist = createPersist(contextId.timestamp)
- def getPersist(timestamp: Long): Persist = {
+ private val persistFactory = SinkFactory(persistParams, name)
+ private val defaultPersist: Sink = createPersist(contextId.timestamp)
+ def getPersist(timestamp: Long): Sink = {
if (timestamp == contextId.timestamp) getPersist()
else createPersist(timestamp)
}
- def getPersist(): Persist = defaultPersist
- private def createPersist(t: Long): Persist = {
+ def getPersist(): Sink = defaultPersist
+ private def createPersist(t: Long): Sink = {
procType match {
case BatchProcessType => persistFactory.getPersists(t, true)
case StreamingProcessType => persistFactory.getPersists(t, false)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/ConsoleSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/ConsoleSink.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/ConsoleSink.scala
new file mode 100644
index 0000000..e2d2d53
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/ConsoleSink.scala
@@ -0,0 +1,84 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import org.apache.griffin.measure.utils.JsonUtil
+import org.apache.griffin.measure.utils.ParamUtil._
+import org.apache.spark.rdd.RDD
+
+/**
+ * persist metric and record to console, for debug
+ */
+case class ConsoleSink(config: Map[String, Any], metricName: String, timeStamp: Long) extends Sink {
+
+ val block: Boolean = true
+
+ val MaxLogLines = "max.log.lines"
+
+ val maxLogLines = config.getInt(MaxLogLines, 100)
+
+ def available(): Boolean = true
+
+ def start(msg: String): Unit = {
+ println(s"[${timeStamp}] ${metricName} start: ${msg}")
+ }
+ def finish(): Unit = {
+ println(s"[${timeStamp}] ${metricName} finish")
+ }
+
+ def log(rt: Long, msg: String): Unit = {
+ println(s"[${timeStamp}] ${rt}: ${msg}")
+ }
+
+ def persistRecords(records: RDD[String], name: String): Unit = {
+// println(s"${metricName} [${timeStamp}] records: ")
+// try {
+// val recordCount = records.count
+// val count = if (maxLogLines < 0) recordCount else scala.math.min(maxLogLines, recordCount)
+// val maxCount = count.toInt
+// if (maxCount > 0) {
+// val recordsArray = records.take(maxCount)
+// recordsArray.foreach(println)
+// }
+// } catch {
+// case e: Throwable => error(e.getMessage)
+// }
+ }
+
+ def persistRecords(records: Iterable[String], name: String): Unit = {
+// println(s"${metricName} [${timeStamp}] records: ")
+// try {
+// val recordCount = records.size
+// val count = if (maxLogLines < 0) recordCount else scala.math.min(maxLogLines, recordCount)
+// if (count > 0) {
+// records.foreach(println)
+// }
+// } catch {
+// case e: Throwable => error(e.getMessage)
+// }
+ }
+
+ def persistMetrics(metrics: Map[String, Any]): Unit = {
+ println(s"${metricName} [${timeStamp}] metrics: ")
+ val json = JsonUtil.toJson(metrics)
+ println(json)
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/HdfsSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/HdfsSink.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/HdfsSink.scala
new file mode 100644
index 0000000..a491fd9
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/HdfsSink.scala
@@ -0,0 +1,186 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import java.util.Date
+
+import org.apache.griffin.measure.utils.{HdfsUtil, JsonUtil}
+import org.apache.griffin.measure.utils.ParamUtil._
+import org.apache.spark.rdd.RDD
+
+/**
+ * persist metric and record to hdfs
+ */
+case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Long) extends Sink {
+
+ val block: Boolean = true
+
+ val Path = "path"
+ val MaxPersistLines = "max.persist.lines"
+ val MaxLinesPerFile = "max.lines.per.file"
+
+ val path = config.getOrElse(Path, "").toString
+ val maxPersistLines = config.getInt(MaxPersistLines, -1)
+ val maxLinesPerFile = math.min(config.getInt(MaxLinesPerFile, 10000), 1000000)
+
+ val StartFile = filePath("_START")
+ val FinishFile = filePath("_FINISH")
+ val MetricsFile = filePath("_METRICS")
+
+ val LogFile = filePath("_LOG")
+
+ var _init = true
+
+ def available(): Boolean = {
+ path.nonEmpty
+ }
+
+ private def logHead: String = {
+ if (_init) {
+ _init = false
+ val dt = new Date(timeStamp)
+ s"================ log of ${dt} ================\n"
+ } else ""
+ }
+
+ private def timeHead(rt: Long): String = {
+ val dt = new Date(rt)
+ s"--- ${dt} ---\n"
+ }
+
+ private def logWrap(rt: Long, msg: String): String = {
+ logHead + timeHead(rt) + s"${msg}\n\n"
+ }
+
+ protected def filePath(file: String): String = {
+ HdfsUtil.getHdfsFilePath(path, s"${metricName}/${timeStamp}/${file}")
+ }
+
+ protected def withSuffix(path: String, suffix: String): String = {
+ s"${path}.${suffix}"
+ }
+
+ def start(msg: String): Unit = {
+ try {
+ HdfsUtil.writeContent(StartFile, msg)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+ def finish(): Unit = {
+ try {
+ HdfsUtil.createEmptyFile(FinishFile)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+ def log(rt: Long, msg: String): Unit = {
+ try {
+ val logStr = logWrap(rt, msg)
+ HdfsUtil.appendContent(LogFile, logStr)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+ private def getHdfsPath(path: String, groupId: Int): String = {
+ HdfsUtil.getHdfsFilePath(path, s"${groupId}")
+ }
+ private def getHdfsPath(path: String, ptnId: Int, groupId: Int): String = {
+ HdfsUtil.getHdfsFilePath(path, s"${ptnId}.${groupId}")
+ }
+
+ private def clearOldRecords(path: String): Unit = {
+ HdfsUtil.deleteHdfsPath(path)
+ }
+
+ def persistRecords(records: RDD[String], name: String): Unit = {
+ val path = filePath(name)
+ clearOldRecords(path)
+ try {
+ val recordCount = records.count
+ val count = if (maxPersistLines < 0) recordCount else scala.math.min(maxPersistLines, recordCount)
+ if (count > 0) {
+ val groupCount = ((count - 1) / maxLinesPerFile + 1).toInt
+ if (groupCount <= 1) {
+ val recs = records.take(count.toInt)
+ persistRecords2Hdfs(path, recs)
+ } else {
+ val groupedRecords: RDD[(Long, Iterable[String])] =
+ records.zipWithIndex.flatMap { r =>
+ val gid = r._2 / maxLinesPerFile
+ if (gid < groupCount) Some((gid, r._1)) else None
+ }.groupByKey()
+ groupedRecords.foreach { group =>
+ val (gid, recs) = group
+ val hdfsPath = if (gid == 0) path else withSuffix(path, gid.toString)
+ persistRecords2Hdfs(hdfsPath, recs)
+ }
+ }
+ }
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+ def persistRecords(records: Iterable[String], name: String): Unit = {
+ val path = filePath(name)
+ clearOldRecords(path)
+ try {
+ val recordCount = records.size
+ val count = if (maxPersistLines < 0) recordCount else scala.math.min(maxPersistLines, recordCount)
+ if (count > 0) {
+ val groupCount = (count - 1) / maxLinesPerFile + 1
+ if (groupCount <= 1) {
+ val recs = records.take(count.toInt)
+ persistRecords2Hdfs(path, recs)
+ } else {
+ val groupedRecords = records.grouped(maxLinesPerFile).zipWithIndex
+ groupedRecords.take(groupCount).foreach { group =>
+ val (recs, gid) = group
+ val hdfsPath = getHdfsPath(path, gid)
+ persistRecords2Hdfs(hdfsPath, recs)
+ }
+ }
+ }
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+ def persistMetrics(metrics: Map[String, Any]): Unit = {
+ try {
+ val json = JsonUtil.toJson(metrics)
+ persistRecords2Hdfs(MetricsFile, json :: Nil)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+ private def persistRecords2Hdfs(hdfsPath: String, records: Iterable[String]): Unit = {
+ try {
+ val recStr = records.mkString("\n")
+ HdfsUtil.writeContent(hdfsPath, recStr)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/HttpSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/HttpSink.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/HttpSink.scala
new file mode 100644
index 0000000..1f5b20c
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/HttpSink.scala
@@ -0,0 +1,82 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import org.apache.griffin.measure.utils.{HttpUtil, JsonUtil, TimeUtil}
+import org.apache.griffin.measure.utils.ParamUtil._
+import org.apache.spark.rdd.RDD
+
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, Future}
+
+/**
+ * persist metric and record through http request
+ */
+case class HttpSink(config: Map[String, Any], metricName: String,
+ timeStamp: Long, block: Boolean
+ ) extends Sink {
+
+ val Api = "api"
+ val Method = "method"
+ val OverTime = "over.time"
+ val Retry = "retry"
+
+ val api = config.getString(Api, "")
+ val method = config.getString(Method, "post")
+ val overTime = TimeUtil.milliseconds(config.getString(OverTime, "")).getOrElse(-1L)
+ val retry = config.getInt(Retry, 10)
+
+ val _Value = "value"
+
+ def available(): Boolean = {
+ api.nonEmpty
+ }
+
+ def start(msg: String): Unit = {}
+ def finish(): Unit = {}
+
+ private def httpResult(dataMap: Map[String, Any]) = {
+ try {
+ val data = JsonUtil.toJson(dataMap)
+ // http request
+ val params = Map[String, Object]()
+ val header = Map[String, Object](("Content-Type","application/json"))
+
+ def func(): (Long, Future[Boolean]) = {
+ import scala.concurrent.ExecutionContext.Implicits.global
+ (timeStamp, Future(HttpUtil.httpRequest(api, method, params, header, data)))
+ }
+ if (block) SinkTaskRunner.addBlockTask(func _, retry, overTime)
+ else SinkTaskRunner.addNonBlockTask(func _, retry)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+
+ }
+
+ def log(rt: Long, msg: String): Unit = {}
+
+ def persistRecords(records: RDD[String], name: String): Unit = {}
+ def persistRecords(records: Iterable[String], name: String): Unit = {}
+
+ def persistMetrics(metrics: Map[String, Any]): Unit = {
+ httpResult(metrics)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/MongoSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/MongoSink.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/MongoSink.scala
new file mode 100644
index 0000000..08360ba
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/MongoSink.scala
@@ -0,0 +1,128 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import org.apache.griffin.measure.utils.ParamUtil._
+import org.apache.griffin.measure.utils.TimeUtil
+import org.apache.spark.rdd.RDD
+import org.mongodb.scala._
+import org.mongodb.scala.model.{Filters, UpdateOptions, Updates}
+import org.mongodb.scala.result.UpdateResult
+
+import scala.concurrent.Future
+
+/**
+ * persist metric and record to mongo
+ */
+case class MongoSink(config: Map[String, Any], metricName: String,
+ timeStamp: Long, block: Boolean
+ ) extends Sink {
+
+ MongoConnection.init(config)
+
+ val OverTime = "over.time"
+ val Retry = "retry"
+
+ val overTime = TimeUtil.milliseconds(config.getString(OverTime, "")).getOrElse(-1L)
+ val retry = config.getInt(Retry, 10)
+
+ val _MetricName = "metricName"
+ val _Timestamp = "timestamp"
+ val _Value = "value"
+
+ def available(): Boolean = MongoConnection.dataConf.available
+
+ def start(msg: String): Unit = {}
+ def finish(): Unit = {}
+
+ def log(rt: Long, msg: String): Unit = {}
+
+ def persistRecords(records: RDD[String], name: String): Unit = {}
+ def persistRecords(records: Iterable[String], name: String): Unit = {}
+
+ def persistMetrics(metrics: Map[String, Any]): Unit = {
+ mongoInsert(metrics)
+ }
+
+ private val filter = Filters.and(
+ Filters.eq(_MetricName, metricName),
+ Filters.eq(_Timestamp, timeStamp)
+ )
+
+ private def mongoInsert(dataMap: Map[String, Any]): Unit = {
+ try {
+ val update = Updates.set(_Value, dataMap)
+ def func(): (Long, Future[UpdateResult]) = {
+ (timeStamp, MongoConnection.getDataCollection.updateOne(
+ filter, update, UpdateOptions().upsert(true)).toFuture)
+ }
+ if (block) SinkTaskRunner.addBlockTask(func _, retry, overTime)
+ else SinkTaskRunner.addNonBlockTask(func _, retry)
+ } catch {
+ case e: Throwable => error(e.getMessage)
+ }
+ }
+
+}
+
+object MongoConnection {
+
+ case class MongoConf(url: String, database: String, collection: String) {
+ def available: Boolean = url.nonEmpty && database.nonEmpty && collection.nonEmpty
+ }
+
+ val _MongoHead = "mongodb://"
+
+ val Url = "url"
+ val Database = "database"
+ val Collection = "collection"
+
+ private var initialed = false
+
+ var dataConf: MongoConf = _
+ private var dataCollection: MongoCollection[Document] = _
+
+ def getDataCollection = dataCollection
+
+ def init(config: Map[String, Any]): Unit = {
+ if (!initialed) {
+ dataConf = mongoConf(config)
+ dataCollection = mongoCollection(dataConf)
+ initialed = true
+ }
+ }
+
+ private def mongoConf(cfg: Map[String, Any]): MongoConf = {
+ val url = cfg.getString(Url, "").trim
+ val mongoUrl = if (url.startsWith(_MongoHead)) url else {
+ _MongoHead + url
+ }
+ MongoConf(
+ mongoUrl,
+ cfg.getString(Database, ""),
+ cfg.getString(Collection, "")
+ )
+ }
+ private def mongoCollection(mongoConf: MongoConf): MongoCollection[Document] = {
+ val mongoClient: MongoClient = MongoClient(mongoConf.url)
+ val database: MongoDatabase = mongoClient.getDatabase(mongoConf.database)
+ database.getCollection(mongoConf.collection)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/MultiSinks.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/MultiSinks.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/MultiSinks.scala
new file mode 100644
index 0000000..a6d36bc
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/MultiSinks.scala
@@ -0,0 +1,85 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import org.apache.spark.rdd.RDD
+
+/**
+ * persist metric and record in multiple ways
+ */
+case class MultiSinks(persists: Iterable[Sink]) extends Sink {
+
+ val block: Boolean = false
+
+ val metricName: String = persists match {
+ case Nil => ""
+ case _ => persists.head.metricName
+ }
+
+ val timeStamp: Long = persists match {
+ case Nil => 0
+ case _ => persists.head.timeStamp
+ }
+
+ val config: Map[String, Any] = Map[String, Any]()
+
+ def available(): Boolean = { persists.exists(_.available()) }
+
+ def start(msg: String): Unit = { persists.foreach(_.start(msg)) }
+ def finish(): Unit = { persists.foreach(_.finish()) }
+
+ def log(rt: Long, msg: String): Unit = {
+ persists.foreach { persist =>
+ try {
+ persist.log(rt, msg)
+ } catch {
+ case e: Throwable => error(s"log error: ${e.getMessage}")
+ }
+ }
+ }
+
+ def persistRecords(records: RDD[String], name: String): Unit = {
+ persists.foreach { persist =>
+ try {
+ persist.persistRecords(records, name)
+ } catch {
+ case e: Throwable => error(s"persist records error: ${e.getMessage}")
+ }
+ }
+ }
+ def persistRecords(records: Iterable[String], name: String): Unit = {
+ persists.foreach { persist =>
+ try {
+ persist.persistRecords(records, name)
+ } catch {
+ case e: Throwable => error(s"persist records error: ${e.getMessage}")
+ }
+ }
+ }
+ def persistMetrics(metrics: Map[String, Any]): Unit = {
+ persists.foreach { persist =>
+ try {
+ persist.persistMetrics(metrics)
+ } catch {
+ case e: Throwable => error(s"persist metrics error: ${e.getMessage}")
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/Sink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/Sink.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/Sink.scala
new file mode 100644
index 0000000..62be4b7
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/Sink.scala
@@ -0,0 +1,47 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import org.apache.griffin.measure.Loggable
+import org.apache.spark.rdd.RDD
+
+/**
+ * persist metric and record
+ */
+trait Sink extends Loggable with Serializable {
+ val metricName: String
+ val timeStamp: Long
+
+ val config: Map[String, Any]
+
+ val block: Boolean
+
+ def available(): Boolean
+
+ def start(msg: String): Unit
+ def finish(): Unit
+
+ def log(rt: Long, msg: String): Unit
+
+ def persistRecords(records: RDD[String], name: String): Unit
+ def persistRecords(records: Iterable[String], name: String): Unit
+
+ def persistMetrics(metrics: Map[String, Any]): Unit
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkFactory.scala
new file mode 100644
index 0000000..57f7309
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkFactory.scala
@@ -0,0 +1,58 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import org.apache.griffin.measure.configuration.params.PersistParam
+
+import scala.util.{Success, Try}
+
+
+case class SinkFactory(persistParams: Iterable[PersistParam], metricName: String) extends Serializable {
+
+ val HDFS_REGEX = """^(?i)hdfs$""".r
+ val HTTP_REGEX = """^(?i)http$""".r
+ val LOG_REGEX = """^(?i)log$""".r
+ val MONGO_REGEX = """^(?i)mongo$""".r
+
+ /**
+ * create persist
+ * @param timeStamp the timestamp of persist
+ * @param block persist write metric in block or non-block way
+ * @return persist
+ */
+ def getPersists(timeStamp: Long, block: Boolean): MultiSinks = {
+ MultiSinks(persistParams.flatMap(param => getPersist(timeStamp, param, block)))
+ }
+
+ private def getPersist(timeStamp: Long, persistParam: PersistParam, block: Boolean): Option[Sink] = {
+ val config = persistParam.getConfig
+ val persistTry = persistParam.getType match {
+ case LOG_REGEX() => Try(ConsoleSink(config, metricName, timeStamp))
+ case HDFS_REGEX() => Try(HdfsSink(config, metricName, timeStamp))
+ case HTTP_REGEX() => Try(HttpSink(config, metricName, timeStamp, block))
+ case MONGO_REGEX() => Try(MongoSink(config, metricName, timeStamp, block))
+ case _ => throw new Exception("not supported persist type")
+ }
+ persistTry match {
+ case Success(persist) if (persist.available) => Some(persist)
+ case _ => None
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkTaskRunner.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkTaskRunner.scala b/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkTaskRunner.scala
new file mode 100644
index 0000000..dacdb06
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/sink/SinkTaskRunner.scala
@@ -0,0 +1,101 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.griffin.measure.context.sink
+
+import java.util.Date
+import java.util.concurrent.TimeUnit
+
+import org.apache.griffin.measure.Loggable
+
+import scala.concurrent.duration._
+import scala.concurrent._
+import scala.util.{Failure, Success}
+
+/**
+ * persist task runner, to persist metrics in block or non-block mode
+ */
+object SinkTaskRunner extends Loggable {
+
+ import scala.concurrent.ExecutionContext.Implicits.global
+
+ val MAX_RETRY = 100
+
+ def addNonBlockTask(func: () => (Long, Future[_]), retry: Int): Unit = {
+ val r = validRetryNum(retry)
+ nonBlockExecute(func, r)
+ }
+
+ def addBlockTask(func: () => (Long, Future[_]), retry: Int, wait: Long): Unit = {
+ val r = validRetryNum(retry)
+ val duration = if (wait >= 0) Duration(wait, TimeUnit.MILLISECONDS) else Duration.Inf
+ blockExecute(func, r, duration)
+ }
+
+ private def nonBlockExecute(func: () => (Long, Future[_]), retry: Int): Unit = {
+ val nextRetry = nextRetryCount(retry)
+ val st = new Date().getTime
+ val (t, res) = func()
+ res.onComplete {
+ case Success(value) => {
+ val et = new Date().getTime
+ info(s"task ${t} success with (${value}) [ using time ${et - st} ms ]")
+ }
+ case Failure(e) => {
+ val et = new Date().getTime
+ warn(s"task ${t} fails [ using time ${et - st} ms ] : ${e.getMessage}")
+ if (nextRetry >= 0) {
+ info(s"task ${t} retry [ rest retry count: ${nextRetry} ]")
+ nonBlockExecute(func, nextRetry)
+ } else {
+ error(s"task fails: task ${t} retry ends but fails")
+ }
+ }
+ }
+ }
+
+ private def blockExecute(func: () => (Long, Future[_]), retry: Int, waitDuration: Duration): Unit = {
+ val nextRetry = nextRetryCount(retry)
+ val st = new Date().getTime
+ val (t, res) = func()
+ try {
+ val value = Await.result(res, waitDuration)
+ val et = new Date().getTime
+ info(s"task ${t} success with (${value}) [ using time ${et - st} ms ]")
+ } catch {
+ case e: Throwable => {
+ val et = new Date().getTime
+ warn(s"task ${t} fails [ using time ${et - st} ms ] : ${e.getMessage}")
+ if (nextRetry >= 0) {
+ info(s"task ${t} retry [ rest retry count: ${nextRetry} ]")
+ blockExecute(func, nextRetry, waitDuration)
+ } else {
+ error(s"task fails: task ${t} retry ends but fails")
+ }
+ }
+ }
+ }
+
+ private def validRetryNum(retry: Int): Int = {
+ if (retry > MAX_RETRY) MAX_RETRY else retry
+ }
+ private def nextRetryCount(retry: Int): Int = {
+ if (retry >= 0) retry - 1 else -1
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/HdfsPersist.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/HdfsPersist.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/HdfsPersist.scala
deleted file mode 100644
index b6cabb3..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/HdfsPersist.scala
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import java.util.Date
-
-import org.apache.griffin.measure.utils.{HdfsUtil, JsonUtil}
-import org.apache.griffin.measure.utils.ParamUtil._
-import org.apache.spark.rdd.RDD
-
-/**
- * persist metric and record to hdfs
- */
-case class HdfsPersist(config: Map[String, Any], metricName: String, timeStamp: Long) extends Persist {
-
- val block: Boolean = true
-
- val Path = "path"
- val MaxPersistLines = "max.persist.lines"
- val MaxLinesPerFile = "max.lines.per.file"
-
- val path = config.getOrElse(Path, "").toString
- val maxPersistLines = config.getInt(MaxPersistLines, -1)
- val maxLinesPerFile = math.min(config.getInt(MaxLinesPerFile, 10000), 1000000)
-
- val StartFile = filePath("_START")
- val FinishFile = filePath("_FINISH")
- val MetricsFile = filePath("_METRICS")
-
- val LogFile = filePath("_LOG")
-
- var _init = true
-
- def available(): Boolean = {
- path.nonEmpty
- }
-
- private def logHead: String = {
- if (_init) {
- _init = false
- val dt = new Date(timeStamp)
- s"================ log of ${dt} ================\n"
- } else ""
- }
-
- private def timeHead(rt: Long): String = {
- val dt = new Date(rt)
- s"--- ${dt} ---\n"
- }
-
- private def logWrap(rt: Long, msg: String): String = {
- logHead + timeHead(rt) + s"${msg}\n\n"
- }
-
- protected def filePath(file: String): String = {
- HdfsUtil.getHdfsFilePath(path, s"${metricName}/${timeStamp}/${file}")
- }
-
- protected def withSuffix(path: String, suffix: String): String = {
- s"${path}.${suffix}"
- }
-
- def start(msg: String): Unit = {
- try {
- HdfsUtil.writeContent(StartFile, msg)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
- def finish(): Unit = {
- try {
- HdfsUtil.createEmptyFile(FinishFile)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
- def log(rt: Long, msg: String): Unit = {
- try {
- val logStr = logWrap(rt, msg)
- HdfsUtil.appendContent(LogFile, logStr)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
- private def getHdfsPath(path: String, groupId: Int): String = {
- HdfsUtil.getHdfsFilePath(path, s"${groupId}")
- }
- private def getHdfsPath(path: String, ptnId: Int, groupId: Int): String = {
- HdfsUtil.getHdfsFilePath(path, s"${ptnId}.${groupId}")
- }
-
- private def clearOldRecords(path: String): Unit = {
- HdfsUtil.deleteHdfsPath(path)
- }
-
- def persistRecords(records: RDD[String], name: String): Unit = {
- val path = filePath(name)
- clearOldRecords(path)
- try {
- val recordCount = records.count
- val count = if (maxPersistLines < 0) recordCount else scala.math.min(maxPersistLines, recordCount)
- if (count > 0) {
- val groupCount = ((count - 1) / maxLinesPerFile + 1).toInt
- if (groupCount <= 1) {
- val recs = records.take(count.toInt)
- persistRecords2Hdfs(path, recs)
- } else {
- val groupedRecords: RDD[(Long, Iterable[String])] =
- records.zipWithIndex.flatMap { r =>
- val gid = r._2 / maxLinesPerFile
- if (gid < groupCount) Some((gid, r._1)) else None
- }.groupByKey()
- groupedRecords.foreach { group =>
- val (gid, recs) = group
- val hdfsPath = if (gid == 0) path else withSuffix(path, gid.toString)
- persistRecords2Hdfs(hdfsPath, recs)
- }
- }
- }
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
- def persistRecords(records: Iterable[String], name: String): Unit = {
- val path = filePath(name)
- clearOldRecords(path)
- try {
- val recordCount = records.size
- val count = if (maxPersistLines < 0) recordCount else scala.math.min(maxPersistLines, recordCount)
- if (count > 0) {
- val groupCount = (count - 1) / maxLinesPerFile + 1
- if (groupCount <= 1) {
- val recs = records.take(count.toInt)
- persistRecords2Hdfs(path, recs)
- } else {
- val groupedRecords = records.grouped(maxLinesPerFile).zipWithIndex
- groupedRecords.take(groupCount).foreach { group =>
- val (recs, gid) = group
- val hdfsPath = getHdfsPath(path, gid)
- persistRecords2Hdfs(hdfsPath, recs)
- }
- }
- }
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
- def persistMetrics(metrics: Map[String, Any]): Unit = {
- try {
- val json = JsonUtil.toJson(metrics)
- persistRecords2Hdfs(MetricsFile, json :: Nil)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
- private def persistRecords2Hdfs(hdfsPath: String, records: Iterable[String]): Unit = {
- try {
- val recStr = records.mkString("\n")
- HdfsUtil.writeContent(hdfsPath, recStr)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/HttpPersist.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/HttpPersist.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/HttpPersist.scala
deleted file mode 100644
index a072fa5..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/HttpPersist.scala
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import org.apache.griffin.measure.utils.{HttpUtil, JsonUtil, TimeUtil}
-import org.apache.griffin.measure.utils.ParamUtil._
-import org.apache.spark.rdd.RDD
-
-import scala.concurrent.duration.Duration
-import scala.concurrent.{Await, Future}
-
-/**
- * persist metric and record through http request
- */
-case class HttpPersist(config: Map[String, Any], metricName: String,
- timeStamp: Long, block: Boolean
- ) extends Persist {
-
- val Api = "api"
- val Method = "method"
- val OverTime = "over.time"
- val Retry = "retry"
-
- val api = config.getString(Api, "")
- val method = config.getString(Method, "post")
- val overTime = TimeUtil.milliseconds(config.getString(OverTime, "")).getOrElse(-1L)
- val retry = config.getInt(Retry, 10)
-
- val _Value = "value"
-
- def available(): Boolean = {
- api.nonEmpty
- }
-
- def start(msg: String): Unit = {}
- def finish(): Unit = {}
-
- private def httpResult(dataMap: Map[String, Any]) = {
- try {
- val data = JsonUtil.toJson(dataMap)
- // http request
- val params = Map[String, Object]()
- val header = Map[String, Object](("Content-Type","application/json"))
-
- def func(): (Long, Future[Boolean]) = {
- import scala.concurrent.ExecutionContext.Implicits.global
- (timeStamp, Future(HttpUtil.httpRequest(api, method, params, header, data)))
- }
- if (block) PersistTaskRunner.addBlockTask(func _, retry, overTime)
- else PersistTaskRunner.addNonBlockTask(func _, retry)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
-
- }
-
- def log(rt: Long, msg: String): Unit = {}
-
- def persistRecords(records: RDD[String], name: String): Unit = {}
- def persistRecords(records: Iterable[String], name: String): Unit = {}
-
- def persistMetrics(metrics: Map[String, Any]): Unit = {
- httpResult(metrics)
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/LoggerPersist.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/LoggerPersist.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/LoggerPersist.scala
deleted file mode 100644
index eff4c62..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/LoggerPersist.scala
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import org.apache.griffin.measure.utils.JsonUtil
-import org.apache.griffin.measure.utils.ParamUtil._
-import org.apache.spark.rdd.RDD
-
-/**
- * persist metric and record to console, for debug
- */
-case class LoggerPersist(config: Map[String, Any], metricName: String, timeStamp: Long) extends Persist {
-
- val block: Boolean = true
-
- val MaxLogLines = "max.log.lines"
-
- val maxLogLines = config.getInt(MaxLogLines, 100)
-
- def available(): Boolean = true
-
- def start(msg: String): Unit = {
- println(s"[${timeStamp}] ${metricName} start: ${msg}")
- }
- def finish(): Unit = {
- println(s"[${timeStamp}] ${metricName} finish")
- }
-
- def log(rt: Long, msg: String): Unit = {
- println(s"[${timeStamp}] ${rt}: ${msg}")
- }
-
- def persistRecords(records: RDD[String], name: String): Unit = {
-// println(s"${metricName} [${timeStamp}] records: ")
-// try {
-// val recordCount = records.count
-// val count = if (maxLogLines < 0) recordCount else scala.math.min(maxLogLines, recordCount)
-// val maxCount = count.toInt
-// if (maxCount > 0) {
-// val recordsArray = records.take(maxCount)
-// recordsArray.foreach(println)
-// }
-// } catch {
-// case e: Throwable => error(e.getMessage)
-// }
- }
-
- def persistRecords(records: Iterable[String], name: String): Unit = {
-// println(s"${metricName} [${timeStamp}] records: ")
-// try {
-// val recordCount = records.size
-// val count = if (maxLogLines < 0) recordCount else scala.math.min(maxLogLines, recordCount)
-// if (count > 0) {
-// records.foreach(println)
-// }
-// } catch {
-// case e: Throwable => error(e.getMessage)
-// }
- }
-
- def persistMetrics(metrics: Map[String, Any]): Unit = {
- println(s"${metricName} [${timeStamp}] metrics: ")
- val json = JsonUtil.toJson(metrics)
- println(json)
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/MongoPersist.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/MongoPersist.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/MongoPersist.scala
deleted file mode 100644
index 14b86e4..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/MongoPersist.scala
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import org.apache.griffin.measure.utils.ParamUtil._
-import org.apache.griffin.measure.utils.TimeUtil
-import org.apache.spark.rdd.RDD
-import org.mongodb.scala._
-import org.mongodb.scala.model.{Filters, UpdateOptions, Updates}
-import org.mongodb.scala.result.UpdateResult
-
-import scala.concurrent.Future
-
-/**
- * persist metric and record to mongo
- */
-case class MongoPersist(config: Map[String, Any], metricName: String,
- timeStamp: Long, block: Boolean
- ) extends Persist {
-
- MongoConnection.init(config)
-
- val OverTime = "over.time"
- val Retry = "retry"
-
- val overTime = TimeUtil.milliseconds(config.getString(OverTime, "")).getOrElse(-1L)
- val retry = config.getInt(Retry, 10)
-
- val _MetricName = "metricName"
- val _Timestamp = "timestamp"
- val _Value = "value"
-
- def available(): Boolean = MongoConnection.dataConf.available
-
- def start(msg: String): Unit = {}
- def finish(): Unit = {}
-
- def log(rt: Long, msg: String): Unit = {}
-
- def persistRecords(records: RDD[String], name: String): Unit = {}
- def persistRecords(records: Iterable[String], name: String): Unit = {}
-
- def persistMetrics(metrics: Map[String, Any]): Unit = {
- mongoInsert(metrics)
- }
-
- private val filter = Filters.and(
- Filters.eq(_MetricName, metricName),
- Filters.eq(_Timestamp, timeStamp)
- )
-
- private def mongoInsert(dataMap: Map[String, Any]): Unit = {
- try {
- val update = Updates.set(_Value, dataMap)
- def func(): (Long, Future[UpdateResult]) = {
- (timeStamp, MongoConnection.getDataCollection.updateOne(
- filter, update, UpdateOptions().upsert(true)).toFuture)
- }
- if (block) PersistTaskRunner.addBlockTask(func _, retry, overTime)
- else PersistTaskRunner.addNonBlockTask(func _, retry)
- } catch {
- case e: Throwable => error(e.getMessage)
- }
- }
-
-}
-
-object MongoConnection {
-
- case class MongoConf(url: String, database: String, collection: String) {
- def available: Boolean = url.nonEmpty && database.nonEmpty && collection.nonEmpty
- }
-
- val _MongoHead = "mongodb://"
-
- val Url = "url"
- val Database = "database"
- val Collection = "collection"
-
- private var initialed = false
-
- var dataConf: MongoConf = _
- private var dataCollection: MongoCollection[Document] = _
-
- def getDataCollection = dataCollection
-
- def init(config: Map[String, Any]): Unit = {
- if (!initialed) {
- dataConf = mongoConf(config)
- dataCollection = mongoCollection(dataConf)
- initialed = true
- }
- }
-
- private def mongoConf(cfg: Map[String, Any]): MongoConf = {
- val url = cfg.getString(Url, "").trim
- val mongoUrl = if (url.startsWith(_MongoHead)) url else {
- _MongoHead + url
- }
- MongoConf(
- mongoUrl,
- cfg.getString(Database, ""),
- cfg.getString(Collection, "")
- )
- }
- private def mongoCollection(mongoConf: MongoConf): MongoCollection[Document] = {
- val mongoClient: MongoClient = MongoClient(mongoConf.url)
- val database: MongoDatabase = mongoClient.getDatabase(mongoConf.database)
- database.getCollection(mongoConf.collection)
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/MultiPersists.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/MultiPersists.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/MultiPersists.scala
deleted file mode 100644
index 4c7d0f6..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/MultiPersists.scala
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import org.apache.spark.rdd.RDD
-
-/**
- * persist metric and record in multiple ways
- */
-case class MultiPersists(persists: Iterable[Persist]) extends Persist {
-
- val block: Boolean = false
-
- val metricName: String = persists match {
- case Nil => ""
- case _ => persists.head.metricName
- }
-
- val timeStamp: Long = persists match {
- case Nil => 0
- case _ => persists.head.timeStamp
- }
-
- val config: Map[String, Any] = Map[String, Any]()
-
- def available(): Boolean = { persists.exists(_.available()) }
-
- def start(msg: String): Unit = { persists.foreach(_.start(msg)) }
- def finish(): Unit = { persists.foreach(_.finish()) }
-
- def log(rt: Long, msg: String): Unit = {
- persists.foreach { persist =>
- try {
- persist.log(rt, msg)
- } catch {
- case e: Throwable => error(s"log error: ${e.getMessage}")
- }
- }
- }
-
- def persistRecords(records: RDD[String], name: String): Unit = {
- persists.foreach { persist =>
- try {
- persist.persistRecords(records, name)
- } catch {
- case e: Throwable => error(s"persist records error: ${e.getMessage}")
- }
- }
- }
- def persistRecords(records: Iterable[String], name: String): Unit = {
- persists.foreach { persist =>
- try {
- persist.persistRecords(records, name)
- } catch {
- case e: Throwable => error(s"persist records error: ${e.getMessage}")
- }
- }
- }
- def persistMetrics(metrics: Map[String, Any]): Unit = {
- persists.foreach { persist =>
- try {
- persist.persistMetrics(metrics)
- } catch {
- case e: Throwable => error(s"persist metrics error: ${e.getMessage}")
- }
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/Persist.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/Persist.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/Persist.scala
deleted file mode 100644
index 8adbcc3..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/Persist.scala
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import org.apache.griffin.measure.Loggable
-import org.apache.spark.rdd.RDD
-
-/**
- * persist metric and record
- */
-trait Persist extends Loggable with Serializable {
- val metricName: String
- val timeStamp: Long
-
- val config: Map[String, Any]
-
- val block: Boolean
-
- def available(): Boolean
-
- def start(msg: String): Unit
- def finish(): Unit
-
- def log(rt: Long, msg: String): Unit
-
- def persistRecords(records: RDD[String], name: String): Unit
- def persistRecords(records: Iterable[String], name: String): Unit
-
- def persistMetrics(metrics: Map[String, Any]): Unit
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistFactory.scala
deleted file mode 100644
index 0a2649e..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistFactory.scala
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import org.apache.griffin.measure.configuration.params.PersistParam
-
-import scala.util.{Success, Try}
-
-
-case class PersistFactory(persistParams: Iterable[PersistParam], metricName: String) extends Serializable {
-
- val HDFS_REGEX = """^(?i)hdfs$""".r
- val HTTP_REGEX = """^(?i)http$""".r
- val LOG_REGEX = """^(?i)log$""".r
- val MONGO_REGEX = """^(?i)mongo$""".r
-
- /**
- * create persist
- * @param timeStamp the timestamp of persist
- * @param block persist write metric in block or non-block way
- * @return persist
- */
- def getPersists(timeStamp: Long, block: Boolean): MultiPersists = {
- MultiPersists(persistParams.flatMap(param => getPersist(timeStamp, param, block)))
- }
-
- private def getPersist(timeStamp: Long, persistParam: PersistParam, block: Boolean): Option[Persist] = {
- val config = persistParam.getConfig
- val persistTry = persistParam.getType match {
- case LOG_REGEX() => Try(LoggerPersist(config, metricName, timeStamp))
- case HDFS_REGEX() => Try(HdfsPersist(config, metricName, timeStamp))
- case HTTP_REGEX() => Try(HttpPersist(config, metricName, timeStamp, block))
- case MONGO_REGEX() => Try(MongoPersist(config, metricName, timeStamp, block))
- case _ => throw new Exception("not supported persist type")
- }
- persistTry match {
- case Success(persist) if (persist.available) => Some(persist)
- case _ => None
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/23a3e999/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistTaskRunner.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistTaskRunner.scala b/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistTaskRunner.scala
deleted file mode 100644
index af4f36b..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/context/writer/PersistTaskRunner.scala
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.griffin.measure.context.writer
-
-import java.util.Date
-import java.util.concurrent.TimeUnit
-
-import org.apache.griffin.measure.Loggable
-
-import scala.concurrent.duration._
-import scala.concurrent._
-import scala.util.{Failure, Success}
-
-/**
- * persist task runner, to persist metrics in block or non-block mode
- */
-object PersistTaskRunner extends Loggable {
-
- import scala.concurrent.ExecutionContext.Implicits.global
-
- val MAX_RETRY = 100
-
- def addNonBlockTask(func: () => (Long, Future[_]), retry: Int): Unit = {
- val r = validRetryNum(retry)
- nonBlockExecute(func, r)
- }
-
- def addBlockTask(func: () => (Long, Future[_]), retry: Int, wait: Long): Unit = {
- val r = validRetryNum(retry)
- val duration = if (wait >= 0) Duration(wait, TimeUnit.MILLISECONDS) else Duration.Inf
- blockExecute(func, r, duration)
- }
-
- private def nonBlockExecute(func: () => (Long, Future[_]), retry: Int): Unit = {
- val nextRetry = nextRetryCount(retry)
- val st = new Date().getTime
- val (t, res) = func()
- res.onComplete {
- case Success(value) => {
- val et = new Date().getTime
- info(s"task ${t} success with (${value}) [ using time ${et - st} ms ]")
- }
- case Failure(e) => {
- val et = new Date().getTime
- warn(s"task ${t} fails [ using time ${et - st} ms ] : ${e.getMessage}")
- if (nextRetry >= 0) {
- info(s"task ${t} retry [ rest retry count: ${nextRetry} ]")
- nonBlockExecute(func, nextRetry)
- } else {
- error(s"task fails: task ${t} retry ends but fails")
- }
- }
- }
- }
-
- private def blockExecute(func: () => (Long, Future[_]), retry: Int, waitDuration: Duration): Unit = {
- val nextRetry = nextRetryCount(retry)
- val st = new Date().getTime
- val (t, res) = func()
- try {
- val value = Await.result(res, waitDuration)
- val et = new Date().getTime
- info(s"task ${t} success with (${value}) [ using time ${et - st} ms ]")
- } catch {
- case e: Throwable => {
- val et = new Date().getTime
- warn(s"task ${t} fails [ using time ${et - st} ms ] : ${e.getMessage}")
- if (nextRetry >= 0) {
- info(s"task ${t} retry [ rest retry count: ${nextRetry} ]")
- blockExecute(func, nextRetry, waitDuration)
- } else {
- error(s"task fails: task ${t} retry ends but fails")
- }
- }
- }
- }
-
- private def validRetryNum(retry: Int): Int = {
- if (retry > MAX_RETRY) MAX_RETRY else retry
- }
- private def nextRetryCount(retry: Int): Int = {
- if (retry >= 0) retry - 1 else -1
- }
-
-}