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/10/30 07:08:47 UTC
incubator-griffin git commit: [GRIFFIN-208] log exception details
when it makes sense
Repository: incubator-griffin
Updated Branches:
refs/heads/master bcfd0e269 -> 165ef3ded
[GRIFFIN-208] log exception details when it makes sense
Sometimes error messages are not descriptive enough and do not allow
to pinpoint exact issue. Exception stack traces should make it easier
to troubleshoot issues.
Author: Nikolay Sokolov <ch...@gmail.com>
Closes #448 from chemikadze/GRIFFIN-208.
Project: http://git-wip-us.apache.org/repos/asf/incubator-griffin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-griffin/commit/165ef3de
Tree: http://git-wip-us.apache.org/repos/asf/incubator-griffin/tree/165ef3de
Diff: http://git-wip-us.apache.org/repos/asf/incubator-griffin/diff/165ef3de
Branch: refs/heads/master
Commit: 165ef3ded15f77657b996c9a1e7c79ccee0a3fb1
Parents: bcfd0e2
Author: Nikolay Sokolov <ch...@gmail.com>
Authored: Tue Oct 30 15:08:18 2018 +0800
Committer: William Guo <gu...@apache.org>
Committed: Tue Oct 30 15:08:18 2018 +0800
----------------------------------------------------------------------
.../apache/griffin/measure/Application.scala | 22 +++++++++++++-------
.../org/apache/griffin/measure/Loggable.scala | 8 +++++++
.../datasource/connector/DataConnector.scala | 2 +-
.../batch/AvroBatchDataConnector.scala | 2 +-
.../batch/HiveBatchDataConnector.scala | 2 +-
.../batch/TextDirBatchDataConnector.scala | 2 +-
.../org/apache/griffin/measure/job/DQJob.scala | 3 +++
.../apache/griffin/measure/launch/DQApp.scala | 5 ++++-
.../measure/launch/batch/BatchDQApp.scala | 6 ++++--
.../launch/streaming/StreamingDQApp.scala | 3 ++-
.../measure/sink/ElasticSearchSink.scala | 2 +-
.../apache/griffin/measure/sink/HdfsSink.scala | 14 ++++++-------
.../apache/griffin/measure/sink/MongoSink.scala | 2 +-
.../griffin/measure/sink/MultiSinks.scala | 8 +++----
.../griffin/measure/sink/SinkTaskRunner.scala | 4 ++--
.../apache/griffin/measure/step/DQStep.scala | 3 +++
.../apache/griffin/measure/step/SeqDQStep.scala | 3 +++
.../step/builder/GriffinDslDQStepBuilder.scala | 2 +-
.../transform/DataFrameOpsTransformStep.scala | 2 +-
.../step/transform/SparkSqlTransformStep.scala | 2 +-
.../step/write/DataSourceUpdateWriteStep.scala | 2 +-
.../measure/step/write/MetricFlushStep.scala | 2 +-
.../measure/step/write/MetricWriteStep.scala | 2 +-
.../measure/step/write/RecordWriteStep.scala | 2 +-
.../apache/griffin/measure/utils/FSUtil.scala | 2 +-
.../apache/griffin/measure/utils/HdfsUtil.scala | 4 ++--
26 files changed, 70 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/Application.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/Application.scala b/measure/src/main/scala/org/apache/griffin/measure/Application.scala
index 1bac17b..99c6a25 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/Application.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/Application.scala
@@ -51,13 +51,13 @@ object Application extends Loggable {
val envParam = readParamFile[EnvConfig](envParamFile) match {
case Success(p) => p
case Failure(ex) =>
- error(ex.getMessage)
+ error(ex.getMessage, ex)
sys.exit(-2)
}
val dqParam = readParamFile[DQConfig](dqParamFile) match {
case Success(p) => p
case Failure(ex) =>
- error(ex.getMessage)
+ error(ex.getMessage, ex)
sys.exit(-2)
}
val allParam: GriffinConfig = GriffinConfig(envParam, dqParam)
@@ -79,17 +79,19 @@ object Application extends Loggable {
case Success(_) =>
info("process init success")
case Failure(ex) =>
- error(s"process init error: ${ex.getMessage}")
+ error(s"process init error: ${ex.getMessage}", ex)
shutdown
sys.exit(-5)
}
// dq app run
- dqApp.run match {
- case Success(_) =>
- info("process run success")
+ val success = dqApp.run match {
+ case Success(result) =>
+ info("process run result: " + (if (result) "success" else "failed"))
+ result
+
case Failure(ex) =>
- error(s"process run error: ${ex.getMessage}")
+ error(s"process run error: ${ex.getMessage}", ex)
if (dqApp.retryable) {
throw ex
@@ -104,12 +106,16 @@ object Application extends Loggable {
case Success(_) =>
info("process end success")
case Failure(ex) =>
- error(s"process end error: ${ex.getMessage}")
+ error(s"process end error: ${ex.getMessage}", ex)
shutdown
sys.exit(-5)
}
shutdown
+
+ if (!success) {
+ sys.exit(-5)
+ }
}
private def readParamFile[T <: Param](file: String)(implicit m : ClassTag[T]): Try[T] = {
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/Loggable.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/Loggable.scala b/measure/src/main/scala/org/apache/griffin/measure/Loggable.scala
index 961fd25..2e113ab 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/Loggable.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/Loggable.scala
@@ -36,8 +36,16 @@ trait Loggable {
logger.warn(msg)
}
+ protected def warn(msg: String, e: Throwable): Unit = {
+ logger.warn(msg, e)
+ }
+
protected def error(msg: String): Unit = {
logger.error(msg)
}
+ protected def error(msg: String, e: Throwable): Unit = {
+ logger.error(msg, e)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnector.scala
index ae6a18d..121502f 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnector.scala
@@ -92,7 +92,7 @@ trait DataConnector extends Loggable with Serializable {
} catch {
case e: Throwable =>
- error(s"pre-process of data connector [${id}] error: ${e.getMessage}")
+ error(s"pre-process of data connector [${id}] error: ${e.getMessage}", e)
None
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/AvroBatchDataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/AvroBatchDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/AvroBatchDataConnector.scala
index bf71b2c..09c96d5 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/AvroBatchDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/AvroBatchDataConnector.scala
@@ -60,7 +60,7 @@ case class AvroBatchDataConnector(@transient sparkSession: SparkSession,
preDfOpt
} catch {
case e: Throwable =>
- error(s"load avro file ${concreteFileFullPath} fails")
+ error(s"load avro file ${concreteFileFullPath} fails", e)
None
}
val tmsts = readTmst(ms)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/HiveBatchDataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/HiveBatchDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/HiveBatchDataConnector.scala
index 1df3bd7..91ab07d 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/HiveBatchDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/HiveBatchDataConnector.scala
@@ -56,7 +56,7 @@ case class HiveBatchDataConnector(@transient sparkSession: SparkSession,
preDfOpt
} catch {
case e: Throwable =>
- error(s"load hive table ${concreteTableName} fails: ${e.getMessage}")
+ error(s"load hive table ${concreteTableName} fails: ${e.getMessage}", e)
None
}
val tmsts = readTmst(ms)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/TextDirBatchDataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/TextDirBatchDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/TextDirBatchDataConnector.scala
index a7ab02e..e21335e 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/TextDirBatchDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/TextDirBatchDataConnector.scala
@@ -70,7 +70,7 @@ case class TextDirBatchDataConnector(@transient sparkSession: SparkSession,
}
} catch {
case e: Throwable =>
- error(s"load text dir ${dirPath} fails: ${e.getMessage}")
+ error(s"load text dir ${dirPath} fails: ${e.getMessage}", e)
None
}
val tmsts = readTmst(ms)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/job/DQJob.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/job/DQJob.scala b/measure/src/main/scala/org/apache/griffin/measure/job/DQJob.scala
index c943db9..4573755 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/job/DQJob.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/job/DQJob.scala
@@ -23,6 +23,9 @@ import org.apache.griffin.measure.step.DQStep
case class DQJob(dqSteps: Seq[DQStep]) extends Serializable {
+ /**
+ * @return execution success
+ */
def execute(context: DQContext): Boolean = {
dqSteps.foldLeft(true) { (ret, dqStep) =>
ret && dqStep.execute(context)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/launch/DQApp.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/launch/DQApp.scala b/measure/src/main/scala/org/apache/griffin/measure/launch/DQApp.scala
index 92480d8..71ba89d 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/launch/DQApp.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/launch/DQApp.scala
@@ -34,7 +34,10 @@ trait DQApp extends Loggable with Serializable {
def init: Try[_]
- def run: Try[_]
+ /**
+ * @return execution success
+ */
+ def run: Try[Boolean]
def close: Try[_]
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala b/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
index e2dbc8d..97bffdd 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
@@ -64,7 +64,7 @@ case class BatchDQApp(allParam: GriffinConfig) extends DQApp {
GriffinUDFAgent.register(sqlContext)
}
- def run: Try[_] = Try {
+ def run: Try[Boolean] = Try {
// start time
val startTime = new Date().getTime
@@ -88,7 +88,7 @@ case class BatchDQApp(allParam: GriffinConfig) extends DQApp {
val dqJob = DQJobBuilder.buildDQJob(dqContext, dqParam.getEvaluateRule)
// dq job execute
- dqJob.execute(dqContext)
+ val result = dqJob.execute(dqContext)
// end time
val endTime = new Date().getTime
@@ -99,6 +99,8 @@ case class BatchDQApp(allParam: GriffinConfig) extends DQApp {
// finish
dqContext.getSink().finish()
+
+ result
}
def close: Try[_] = Try {
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/launch/streaming/StreamingDQApp.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/launch/streaming/StreamingDQApp.scala b/measure/src/main/scala/org/apache/griffin/measure/launch/streaming/StreamingDQApp.scala
index eb31a5e..0de8980 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/launch/streaming/StreamingDQApp.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/launch/streaming/StreamingDQApp.scala
@@ -77,7 +77,7 @@ case class StreamingDQApp(allParam: GriffinConfig) extends DQApp {
GriffinUDFAgent.register(sqlContext)
}
- def run: Try[_] = Try {
+ def run: Try[Boolean] = Try {
// streaming context
val ssc = StreamingContext.getOrCreate(sparkParam.getCpDir, () => {
@@ -127,6 +127,7 @@ case class StreamingDQApp(allParam: GriffinConfig) extends DQApp {
// finish
globalContext.getSink().finish()
+ true
}
def close: Try[_] = Try {
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
index 3c20a0e..745f760 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
@@ -69,7 +69,7 @@ case class ElasticSearchSink(config: Map[String, Any], metricName: String,
if (block) SinkTaskRunner.addBlockTask(func _, retry, connectionTimeout)
else SinkTaskRunner.addNonBlockTask(func _, retry)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/sink/HdfsSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/HdfsSink.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/HdfsSink.scala
index 588fabf..d103b32 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/HdfsSink.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/HdfsSink.scala
@@ -81,7 +81,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
try {
HdfsUtil.writeContent(StartFile, msg)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
@@ -89,7 +89,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
try {
HdfsUtil.createEmptyFile(FinishFile)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
@@ -98,7 +98,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
val logStr = logWrap(rt, msg)
HdfsUtil.appendContent(LogFile, logStr)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
@@ -142,7 +142,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
}
}
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
@@ -170,7 +170,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
}
}
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
@@ -179,7 +179,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
val json = JsonUtil.toJson(metrics)
sinkRecords2Hdfs(MetricsFile, json :: Nil)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
@@ -188,7 +188,7 @@ case class HdfsSink(config: Map[String, Any], metricName: String, timeStamp: Lon
val recStr = records.mkString("\n")
HdfsUtil.writeContent(hdfsPath, recStr)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/sink/MongoSink.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/MongoSink.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/MongoSink.scala
index ab59e59..c090201 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/MongoSink.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/MongoSink.scala
@@ -77,7 +77,7 @@ case class MongoSink(config: Map[String, Any], metricName: String,
if (block) SinkTaskRunner.addBlockTask(func _, retry, overTime)
else SinkTaskRunner.addNonBlockTask(func _, retry)
} catch {
- case e: Throwable => error(e.getMessage)
+ case e: Throwable => error(e.getMessage, e)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/sink/MultiSinks.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/MultiSinks.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/MultiSinks.scala
index cca1ff9..b9f72da 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/MultiSinks.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/MultiSinks.scala
@@ -45,7 +45,7 @@ case class MultiSinks(sinks: Iterable[Sink]) extends Sink {
try {
sink.log(rt, msg)
} catch {
- case e: Throwable => error(s"log error: ${e.getMessage}")
+ case e: Throwable => error(s"log error: ${e.getMessage}", e)
}
}
}
@@ -55,7 +55,7 @@ case class MultiSinks(sinks: Iterable[Sink]) extends Sink {
try {
sink.sinkRecords(records, name)
} catch {
- case e: Throwable => error(s"sink records error: ${e.getMessage}")
+ case e: Throwable => error(s"sink records error: ${e.getMessage}", e)
}
}
}
@@ -64,7 +64,7 @@ case class MultiSinks(sinks: Iterable[Sink]) extends Sink {
try {
sink.sinkRecords(records, name)
} catch {
- case e: Throwable => error(s"sink records error: ${e.getMessage}")
+ case e: Throwable => error(s"sink records error: ${e.getMessage}", e)
}
}
}
@@ -73,7 +73,7 @@ case class MultiSinks(sinks: Iterable[Sink]) extends Sink {
try {
sink.sinkMetrics(metrics)
} catch {
- case e: Throwable => error(s"sink metrics error: ${e.getMessage}")
+ case e: Throwable => error(s"sink metrics error: ${e.getMessage}", e)
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/sink/SinkTaskRunner.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/SinkTaskRunner.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/SinkTaskRunner.scala
index ca38629..9584455 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/SinkTaskRunner.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/SinkTaskRunner.scala
@@ -64,7 +64,7 @@ object SinkTaskRunner extends Loggable {
info(s"task ${t} retry [ rest retry count: ${nextRetry} ]")
nonBlockExecute(func, nextRetry)
} else {
- error(s"task fails: task ${t} retry ends but fails")
+ error(s"task fails: task ${t} retry ends but fails", e)
}
}
}
@@ -86,7 +86,7 @@ object SinkTaskRunner extends Loggable {
info(s"task ${t} retry [ rest retry count: ${nextRetry} ]")
blockExecute(func, nextRetry, waitDuration)
} else {
- error(s"task fails: task ${t} retry ends but fails")
+ error(s"task fails: task ${t} retry ends but fails", e)
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/DQStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/DQStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/DQStep.scala
index 8a34118..60c8477 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/DQStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/DQStep.scala
@@ -25,6 +25,9 @@ trait DQStep extends Loggable {
val name: String
+ /**
+ * @return execution success
+ */
def execute(context: DQContext): Boolean
def getNames(): Seq[String] = name :: Nil
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/SeqDQStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/SeqDQStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/SeqDQStep.scala
index b827604..83e4df7 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/SeqDQStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/SeqDQStep.scala
@@ -29,6 +29,9 @@ case class SeqDQStep(dqSteps: Seq[DQStep]) extends DQStep {
val rule: String = ""
val details: Map[String, Any] = Map()
+ /**
+ * @return execution success
+ */
def execute(context: DQContext): Boolean = {
dqSteps.foldLeft(true) { (ret, dqStep) =>
ret && dqStep.execute(context)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/builder/GriffinDslDQStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/GriffinDslDQStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/GriffinDslDQStepBuilder.scala
index aa43cf6..a2ad536 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/GriffinDslDQStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/GriffinDslDQStepBuilder.scala
@@ -50,7 +50,7 @@ case class GriffinDslDQStepBuilder(dataSourceNames: Seq[String],
}
} catch {
case e: Throwable =>
- error(s"generate rule plan ${name} fails: ${e.getMessage}")
+ error(s"generate rule plan ${name} fails: ${e.getMessage}", e)
Nil
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
index a2bf46e..4ac35b2 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
@@ -46,7 +46,7 @@ case class DataFrameOpsTransformStep(name: String,
true
} catch {
case e: Throwable =>
- error(s"run data frame ops [ ${rule} ] error: ${e.getMessage}")
+ error(s"run data frame ops [ ${rule} ] error: ${e.getMessage}", e)
false
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
index ca03f79..39b6a0e 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
@@ -38,7 +38,7 @@ case class SparkSqlTransformStep(name: String,
true
} catch {
case e: Throwable =>
- error(s"run spark sql [ ${rule} ] error: ${e.getMessage}")
+ error(s"run spark sql [ ${rule} ] error: ${e.getMessage}", e)
false
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/write/DataSourceUpdateWriteStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/write/DataSourceUpdateWriteStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/write/DataSourceUpdateWriteStep.scala
index f34e003..d2805cf 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/write/DataSourceUpdateWriteStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/write/DataSourceUpdateWriteStep.scala
@@ -51,7 +51,7 @@ case class DataSourceUpdateWriteStep(dsName: String,
Some(df)
} catch {
case e: Throwable =>
- error(s"get data frame ${name} fails")
+ error(s"get data frame ${name} fails", e)
None
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
index e787d96..09af3cf 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
@@ -37,7 +37,7 @@ case class MetricFlushStep() extends WriteStep {
true
} catch {
case e: Throwable =>
- error(s"flush metrics error: ${e.getMessage}")
+ error(s"flush metrics error: ${e.getMessage}", e)
false
}
ret && pr
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricWriteStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricWriteStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricWriteStep.scala
index bc721f2..c7ebae7 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricWriteStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricWriteStep.scala
@@ -89,7 +89,7 @@ case class MetricWriteStep(name: String,
} else Nil
} catch {
case e: Throwable =>
- error(s"get metric ${name} fails")
+ error(s"get metric ${name} fails", e)
Nil
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/step/write/RecordWriteStep.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/write/RecordWriteStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/write/RecordWriteStep.scala
index 1fef694..3dcbe90 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/write/RecordWriteStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/write/RecordWriteStep.scala
@@ -80,7 +80,7 @@ case class RecordWriteStep(name: String,
Some(df)
} catch {
case e: Throwable =>
- error(s"get data frame ${name} fails")
+ error(s"get data frame ${name} fails", e)
None
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala b/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
index 7ac8b4f..3f945b0 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
@@ -43,7 +43,7 @@ object FSUtil extends Loggable {
FileSystem.get(uri, getConfiguration)
} catch {
case e: Throwable =>
- error(s"get file system error: ${e.getMessage}")
+ error(s"get file system error: ${e.getMessage}", e)
throw e
}
fsMap += (uri.getScheme -> fs)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/165ef3de/measure/src/main/scala/org/apache/griffin/measure/utils/HdfsUtil.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/utils/HdfsUtil.scala b/measure/src/main/scala/org/apache/griffin/measure/utils/HdfsUtil.scala
index 0cae5bd..d23dd46 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/utils/HdfsUtil.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/utils/HdfsUtil.scala
@@ -87,7 +87,7 @@ object HdfsUtil extends Loggable {
implicit val path = new Path(dirPath)
if (getFS.exists(path)) getFS.delete(path, true)
} catch {
- case e: Throwable => error(s"delete path [${dirPath}] error: ${e.getMessage}")
+ case e: Throwable => error(s"delete path [${dirPath}] error: ${e.getMessage}", e)
}
}
@@ -110,7 +110,7 @@ object HdfsUtil extends Loggable {
}
} catch {
case e: Throwable =>
- warn(s"list path [${dirPath}] warn: ${e.getMessage}")
+ warn(s"list path [${dirPath}] warn: ${e.getMessage}", e)
Nil
}
} else Nil