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/15 04:25:23 UTC
[2/2] incubator-griffin git commit: refactor params as dq definition
refactor params as dq definition
Author: Grant <gr...@gmail.com>
Closes #305 from grant-xuexu/refactoring/dqdefinition.
Project: http://git-wip-us.apache.org/repos/asf/incubator-griffin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-griffin/commit/b9e0743f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-griffin/tree/b9e0743f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-griffin/diff/b9e0743f
Branch: refs/heads/master
Commit: b9e0743ff7dd472b93be797c8378cbb5f37a2f61
Parents: 75b7941
Author: Grant <gr...@gmail.com>
Authored: Fri Jun 15 12:25:14 2018 +0800
Committer: Lionel Liu <bh...@163.com>
Committed: Fri Jun 15 12:25:14 2018 +0800
----------------------------------------------------------------------
.../apache/griffin/measure/Application.scala | 4 +-
.../configuration/dqdefinition/DQConfig.scala | 217 +++++++++++++++++++
.../configuration/dqdefinition/EnvConfig.scala | 112 ++++++++++
.../dqdefinition/GriffinConfig.scala | 42 ++++
.../configuration/dqdefinition/Param.scala | 28 +++
.../dqdefinition/reader/ParamFileReader.scala | 44 ++++
.../dqdefinition/reader/ParamJsonReader.scala | 40 ++++
.../dqdefinition/reader/ParamReader.scala | 47 ++++
.../reader/ParamReaderFactory.scala | 48 ++++
.../measure/configuration/params/DQConfig.scala | 217 -------------------
.../configuration/params/EnvConfig.scala | 112 ----------
.../configuration/params/GriffinConfig.scala | 42 ----
.../measure/configuration/params/Param.scala | 28 ---
.../params/reader/ParamFileReader.scala | 43 ----
.../params/reader/ParamJsonReader.scala | 39 ----
.../params/reader/ParamReader.scala | 45 ----
.../params/reader/ParamReaderFactory.scala | 48 ----
.../griffin/measure/context/DQContext.scala | 2 +-
.../streaming/offset/OffsetCacheClient.scala | 2 +-
.../streaming/offset/OffsetCacheFactory.scala | 2 +-
.../griffin/measure/datasource/DataSource.scala | 2 +-
.../measure/datasource/DataSourceFactory.scala | 2 +-
.../datasource/connector/DataConnector.scala | 2 +-
.../connector/DataConnectorFactory.scala | 2 +-
.../batch/AvroBatchDataConnector.scala | 2 +-
.../batch/HiveBatchDataConnector.scala | 2 +-
.../batch/TextDirBatchDataConnector.scala | 2 +-
.../KafkaStreamingStringDataConnector.scala | 2 +-
.../measure/job/builder/DQJobBuilder.scala | 2 +-
.../apache/griffin/measure/launch/DQApp.scala | 2 +-
.../measure/launch/batch/BatchDQApp.scala | 2 +-
.../launch/streaming/StreamingDQApp.scala | 2 +-
.../griffin/measure/sink/SinkFactory.scala | 2 +-
.../builder/BatchDataSourceStepBuilder.scala | 2 +-
.../measure/step/builder/DQStepBuilder.scala | 2 +-
.../builder/DataFrameOpsDQStepBuilder.scala | 2 +-
.../builder/DataSourceParamStepBuilder.scala | 2 +-
.../step/builder/GriffinDslDQStepBuilder.scala | 2 +-
.../step/builder/RuleParamStepBuilder.scala | 2 +-
.../step/builder/SparkSqlDQStepBuilder.scala | 2 +-
.../StreamingDataSourceStepBuilder.scala | 2 +-
.../dsl/transform/AccuracyExpr2DQSteps.scala | 2 +-
.../transform/CompletenessExpr2DQSteps.scala | 2 +-
.../transform/DistinctnessExpr2DQSteps.scala | 2 +-
.../builder/dsl/transform/Expr2DQSteps.scala | 2 +-
.../dsl/transform/ProfilingExpr2DQSteps.scala | 2 +-
.../dsl/transform/TimelinessExpr2DQSteps.scala | 2 +-
.../dsl/transform/UniquenessExpr2DQSteps.scala | 2 +-
.../preproc/PreProcRuleParamGenerator.scala | 2 +-
.../reader/ParamFileReaderSpec.scala | 55 +++++
.../reader/ParamJsonReaderSpec.scala | 65 ++++++
.../params/reader/ParamFileReaderSpec.scala | 55 -----
.../params/reader/ParamJsonReaderSpec.scala | 65 ------
53 files changed, 732 insertions(+), 728 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 566e8df..26891e6 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/Application.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/Application.scala
@@ -19,8 +19,8 @@ under the License.
package org.apache.griffin.measure
import org.apache.griffin.measure.configuration.enums._
-import org.apache.griffin.measure.configuration.params.reader.ParamReaderFactory
-import org.apache.griffin.measure.configuration.params.{GriffinConfig, DQConfig, EnvConfig, Param}
+import org.apache.griffin.measure.configuration.dqdefinition.reader.ParamReaderFactory
+import org.apache.griffin.measure.configuration.dqdefinition.{GriffinConfig, DQConfig, EnvConfig, Param}
import org.apache.griffin.measure.launch.DQApp
import org.apache.griffin.measure.launch.batch.BatchDQApp
import org.apache.griffin.measure.launch.streaming.StreamingDQApp
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala
new file mode 100644
index 0000000..222f6e2
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala
@@ -0,0 +1,217 @@
+/*
+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.configuration.dqdefinition
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty}
+import org.apache.commons.lang.StringUtils
+import org.apache.griffin.measure.configuration.enums._
+
+/**
+ * dq param
+ * @param name name of dq measurement (must)
+ * @param timestamp default timestamp of measure in batch mode (optional)
+ * @param procType batch mode or streaming mode (must)
+ * @param dataSources data sources (must)
+ * @param evaluateRule dq measurement (must)
+ */
+@JsonInclude(Include.NON_NULL)
+case class DQConfig(@JsonProperty("name") name: String,
+ @JsonProperty("timestamp") timestamp: Long,
+ @JsonProperty("process.type") procType: String,
+ @JsonProperty("data.sources") dataSources: List[DataSourceParam],
+ @JsonProperty("evaluate.rule") evaluateRule: EvaluateRuleParam
+ ) extends Param {
+ def getName: String = name
+ def getTimestampOpt: Option[Long] = if (timestamp != null) Some(timestamp) else None
+ def getProcType: String = procType
+ def getDataSources: Seq[DataSourceParam] = {
+ dataSources.foldLeft((Nil: Seq[DataSourceParam], Set[String]())) { (ret, ds) =>
+ val (seq, names) = ret
+ if (!names.contains(ds.getName)){
+ (seq :+ ds, names + ds.getName)
+ } else ret
+ }._1
+ }
+ def getEvaluateRule: EvaluateRuleParam = evaluateRule
+
+ def validate(): Unit = {
+ assert(StringUtils.isNotBlank(name), "dq config name should not be blank")
+ assert(StringUtils.isNotBlank(procType), "process.type should not be blank")
+ assert((dataSources != null), "data.sources should not be null")
+ assert((evaluateRule != null), "evaluate.rule should not be null")
+ getDataSources.foreach(_.validate)
+ evaluateRule.validate
+ }
+}
+
+/**
+ * data source param
+ * @param name data source name (must)
+ * @param connectors data connectors (optional)
+ * @param cache data source cache configuration (must in streaming mode with streaming connectors)
+ */
+@JsonInclude(Include.NON_NULL)
+case class DataSourceParam( @JsonProperty("name") name: String,
+ @JsonProperty("connectors") connectors: List[DataConnectorParam],
+ @JsonProperty("cache") cache: Map[String, Any]
+ ) extends Param {
+ def getName: String = name
+ def getConnectors: Seq[DataConnectorParam] = if (connectors != null) connectors else Nil
+ def getCacheOpt: Option[Map[String, Any]] = if (cache != null) Some(cache) else None
+
+ def validate(): Unit = {
+ assert(StringUtils.isNotBlank(name), "data source name should not be empty")
+ getConnectors.foreach(_.validate)
+ }
+}
+
+/**
+ * data connector param
+ * @param conType data connector type, e.g.: hive, avro, kafka (must)
+ * @param version data connector type version (optional)
+ * @param config detail configuration of data connector (must)
+ * @param preProc pre-process rules after load data (optional)
+ */
+@JsonInclude(Include.NON_NULL)
+case class DataConnectorParam( @JsonProperty("type") conType: String,
+ @JsonProperty("version") version: String,
+ @JsonProperty("config") config: Map[String, Any],
+ @JsonProperty("pre.proc") preProc: List[RuleParam]
+ ) extends Param {
+ def getType: String = conType
+ def getVersion: String = version
+ def getConfig: Map[String, Any] = if (config != null) config else Map[String, Any]()
+ def getPreProcRules: Seq[RuleParam] = if (preProc != null) preProc else Nil
+
+ def validate(): Unit = {
+ assert(StringUtils.isNotBlank(conType), "data connector type should not be empty")
+ getPreProcRules.foreach(_.validate)
+ }
+}
+
+/**
+ * evaluate rule param
+ * @param rules rules to define dq measurement (optional)
+ */
+@JsonInclude(Include.NON_NULL)
+case class EvaluateRuleParam( @JsonProperty("rules") rules: List[RuleParam]
+ ) extends Param {
+ def getRules: Seq[RuleParam] = if (rules != null) rules else Nil
+
+ def validate(): Unit = {
+ getRules.foreach(_.validate)
+ }
+}
+
+/**
+ * rule param
+ * @param dslType dsl type of this rule (must)
+ * @param dqType dq type of this rule (must if dsl type is "griffin-dsl")
+ * @param name name of result calculated by this rule (must if for later usage)
+ * @param rule rule to define dq step calculation (must)
+ * @param details detail config of rule (optional)
+ * @param cache cache the result for multiple usage (optional, valid for "spark-sql" and "df-opr" mode)
+ * @param metric config for metric output (optional)
+ * @param record config for record output (optional)
+ * @param dsCacheUpdate config for data source cache update output (optional, valid in streaming mode)
+ */
+@JsonInclude(Include.NON_NULL)
+case class RuleParam( @JsonProperty("dsl.type") dslType: String,
+ @JsonProperty("dq.type") dqType: String,
+ @JsonProperty("name") name: String,
+ @JsonProperty("rule") rule: String,
+ @JsonProperty("details") details: Map[String, Any],
+ @JsonProperty("cache") cache: Boolean,
+ @JsonProperty("metric") metric: RuleMetricParam,
+ @JsonProperty("record") record: RuleRecordParam,
+ @JsonProperty("ds.cache.update") dsCacheUpdate: RuleDsCacheUpdateParam
+ ) extends Param {
+ def getDslType: DslType = if (dslType != null) DslType(dslType) else DslType("")
+ def getDqType: DqType = if (dqType != null) DqType(dqType) else DqType("")
+ def getCache: Boolean = if (cache != null) cache else false
+
+ def getName: String = if (name != null) name else ""
+ def getRule: String = if (rule != null) rule else ""
+ def getDetails: Map[String, Any] = if (details != null) details else Map[String, Any]()
+
+ def getMetricOpt: Option[RuleMetricParam] = if (metric != null) Some(metric) else None
+ def getRecordOpt: Option[RuleRecordParam] = if (record != null) Some(record) else None
+ def getDsCacheUpdateOpt: Option[RuleDsCacheUpdateParam] = if (dsCacheUpdate != null) Some(dsCacheUpdate) else None
+
+ def replaceName(newName: String): RuleParam = {
+ if (StringUtils.equals(newName, name)) this
+ else RuleParam(dslType, dqType, newName, rule, details, cache, metric, record, dsCacheUpdate)
+ }
+ def replaceRule(newRule: String): RuleParam = {
+ if (StringUtils.equals(newRule, rule)) this
+ else RuleParam(dslType, dqType, name, newRule, details, cache, metric, record, dsCacheUpdate)
+ }
+ def replaceDetails(newDetails: Map[String, Any]): RuleParam = {
+ RuleParam(dslType, dqType, name, rule, newDetails, cache, metric, record, dsCacheUpdate)
+ }
+
+ def validate(): Unit = {
+ assert(!(getDslType.equals(GriffinDslType) && getDqType.equals(UnknownType)),
+ "unknown dq type for griffin dsl")
+
+ getMetricOpt.foreach(_.validate)
+ getRecordOpt.foreach(_.validate)
+ getDsCacheUpdateOpt.foreach(_.validate)
+ }
+}
+
+/**
+ * metric param of rule
+ * @param name name of metric to output (optional)
+ * @param collectType the normalize strategy to collect metric (optional)
+ */
+@JsonInclude(Include.NON_NULL)
+case class RuleMetricParam( @JsonProperty("name") name: String,
+ @JsonProperty("collect.type") collectType: String
+ ) extends Param {
+ def getNameOpt: Option[String] = if (StringUtils.isNotBlank(name)) Some(name) else None
+ def getCollectType: NormalizeType = if (StringUtils.isNotBlank(collectType)) NormalizeType(collectType) else NormalizeType("")
+
+ def validate(): Unit = {}
+}
+
+/**
+ * record param of rule
+ * @param name name of record to output (optional)
+ */
+@JsonInclude(Include.NON_NULL)
+case class RuleRecordParam( @JsonProperty("name") name: String
+ ) extends Param {
+ def getNameOpt: Option[String] = if (StringUtils.isNotBlank(name)) Some(name) else None
+
+ def validate(): Unit = {}
+}
+
+/**
+ * data source cache update param of rule
+ * @param dsName name of data source to be updated by thie rule result (must)
+ */
+@JsonInclude(Include.NON_NULL)
+case class RuleDsCacheUpdateParam( @JsonProperty("ds.name") dsName: String
+ ) extends Param {
+ def getDsNameOpt: Option[String] = if (StringUtils.isNotBlank(dsName)) Some(dsName) else None
+
+ def validate(): Unit = {}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/EnvConfig.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/EnvConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/EnvConfig.scala
new file mode 100644
index 0000000..54a1a51
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/EnvConfig.scala
@@ -0,0 +1,112 @@
+/*
+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.configuration.dqdefinition
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty}
+import org.apache.commons.lang.StringUtils
+import org.apache.griffin.measure.utils.TimeUtil
+
+/**
+ * environment param
+ * @param sparkParam config of spark environment (must)
+ * @param persistParams config of persist ways (optional)
+ * @param offsetCacheParams config of information cache ways (required in streaming mode)
+ */
+@JsonInclude(Include.NON_NULL)
+case class EnvConfig(@JsonProperty("spark") sparkParam: SparkParam,
+ @JsonProperty("persist") persistParams: List[PersistParam],
+ @JsonProperty("info.cache") offsetCacheParams: List[OffsetCacheParam]
+ ) extends Param {
+ def getSparkParam: SparkParam = sparkParam
+ def getPersistParams: Seq[PersistParam] = if (persistParams != null) persistParams else Nil
+ def getOffsetCacheParams: Seq[OffsetCacheParam] = if (offsetCacheParams != null) offsetCacheParams else Nil
+
+ def validate(): Unit = {
+ assert((sparkParam != null), "spark param should not be null")
+ sparkParam.validate
+ getPersistParams.foreach(_.validate)
+ getOffsetCacheParams.foreach(_.validate)
+ }
+}
+
+/**
+ * spark param
+ * @param logLevel log level of spark application (optional)
+ * @param cpDir checkpoint directory for spark streaming (required in streaming mode)
+ * @param batchInterval batch interval for spark streaming (required in streaming mode)
+ * @param processInterval process interval for streaming dq calculation (required in streaming mode)
+ * @param config extra config for spark environment (optional)
+ * @param initClear clear checkpoint directory or not when initial (optional)
+ */
+@JsonInclude(Include.NON_NULL)
+case class SparkParam( @JsonProperty("log.level") logLevel: String,
+ @JsonProperty("checkpoint.dir") cpDir: String,
+ @JsonProperty("batch.interval") batchInterval: String,
+ @JsonProperty("process.interval") processInterval: String,
+ @JsonProperty("config") config: Map[String, String],
+ @JsonProperty("init.clear") initClear: Boolean
+ ) extends Param {
+ def getLogLevel: String = if (logLevel != null) logLevel else "WARN"
+ def getCpDir: String = if (cpDir != null) cpDir else ""
+ def getBatchInterval: String = if (batchInterval != null) batchInterval else ""
+ def getProcessInterval: String = if (processInterval != null) processInterval else ""
+ def getConfig: Map[String, String] = if (config != null) config else Map[String, String]()
+ def needInitClear: Boolean = if (initClear != null) initClear else false
+
+ def validate(): Unit = {
+// assert(StringUtils.isNotBlank(cpDir), "checkpoint.dir should not be empty")
+// assert(TimeUtil.milliseconds(getBatchInterval).nonEmpty, "batch.interval should be valid time string")
+// assert(TimeUtil.milliseconds(getProcessInterval).nonEmpty, "process.interval should be valid time string")
+ }
+}
+
+/**
+ * persist param
+ * @param persistType persist type, e.g.: log, hdfs, http, mongo (must)
+ * @param config config of persist way (must)
+ */
+@JsonInclude(Include.NON_NULL)
+case class PersistParam( @JsonProperty("type") persistType: String,
+ @JsonProperty("config") config: Map[String, Any]
+ ) extends Param {
+ def getType: String = persistType
+ def getConfig: Map[String, Any] = if (config != null) config else Map[String, Any]()
+
+ def validate(): Unit = {
+ assert(StringUtils.isNotBlank(persistType), "persist type should not be empty")
+ }
+}
+
+/**
+ * offset cache param
+ * @param cacheType offset cache type, e.g.: zookeeper (must)
+ * @param config config of cache way
+ */
+@JsonInclude(Include.NON_NULL)
+case class OffsetCacheParam(@JsonProperty("type") cacheType: String,
+ @JsonProperty("config") config: Map[String, Any]
+ ) extends Param {
+ def getType: String = cacheType
+ def getConfig: Map[String, Any] = if (config != null) config else Map[String, Any]()
+
+ def validate(): Unit = {
+ assert(StringUtils.isNotBlank(cacheType), "info cache type should not be empty")
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/GriffinConfig.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/GriffinConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/GriffinConfig.scala
new file mode 100644
index 0000000..4527d4b
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/GriffinConfig.scala
@@ -0,0 +1,42 @@
+/*
+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.configuration.dqdefinition
+
+import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty}
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+
+/**
+ * full set of griffin configuration
+ * @param envConfig environment configuration (must)
+ * @param dqConfig dq measurement configuration (must)
+ */
+@JsonInclude(Include.NON_NULL)
+case class GriffinConfig(@JsonProperty("env") envConfig: EnvConfig,
+ @JsonProperty("dq") dqConfig: DQConfig
+ ) extends Param {
+ def getEnvConfig: EnvConfig = envConfig
+ def getDqConfig: DQConfig = dqConfig
+
+ def validate(): Unit = {
+ assert((envConfig != null), "environment config should not be null")
+ assert((dqConfig != null), "dq config should not be null")
+ envConfig.validate
+ dqConfig.validate
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/Param.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/Param.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/Param.scala
new file mode 100644
index 0000000..348184e
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/Param.scala
@@ -0,0 +1,28 @@
+/*
+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.configuration.dqdefinition
+
+trait Param extends Serializable {
+
+ /**
+ * validate param internally
+ */
+ def validate(): Unit
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReader.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReader.scala
new file mode 100644
index 0000000..c69eb72
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReader.scala
@@ -0,0 +1,44 @@
+/*
+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.configuration.dqdefinition.reader
+
+import org.apache.griffin.measure.configuration.dqdefinition.Param
+import org.apache.griffin.measure.utils.{HdfsUtil, JsonUtil}
+
+import scala.util.Try
+
+/**
+ * read params from config file path
+ *
+ * @param filePath: hdfs path ("hdfs://cluster-name/path")
+ * local file path ("file:///path")
+ * relative file path ("relative/path")
+ */
+case class ParamFileReader(filePath: String) extends ParamReader {
+
+ def readConfig[T <: Param](implicit m : Manifest[T]): Try[T] = {
+ Try {
+ val source = HdfsUtil.openFile(filePath)
+ val param = JsonUtil.fromJson[T](source)
+ source.close
+ validate(param)
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReader.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReader.scala
new file mode 100644
index 0000000..da5c057
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReader.scala
@@ -0,0 +1,40 @@
+/*
+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.configuration.dqdefinition.reader
+
+import org.apache.griffin.measure.configuration.dqdefinition.Param
+import org.apache.griffin.measure.utils.JsonUtil
+
+import scala.util.Try
+
+/**
+ * read params from json string directly
+ *
+ * @param jsonString
+ */
+case class ParamJsonReader(jsonString: String) extends ParamReader {
+
+ def readConfig[T <: Param](implicit m : Manifest[T]): Try[T] = {
+ Try {
+ val param = JsonUtil.fromJson[T](jsonString)
+ validate(param)
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReader.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReader.scala
new file mode 100644
index 0000000..ec47cb0
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReader.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.configuration.dqdefinition.reader
+
+import org.apache.griffin.measure.Loggable
+import org.apache.griffin.measure.configuration.dqdefinition.Param
+
+import scala.util.Try
+
+trait ParamReader extends Loggable with Serializable {
+
+ /**
+ * read config param
+ *
+ * @tparam T param type expected
+ * @return parsed param
+ */
+ def readConfig[T <: Param](implicit m : Manifest[T]): Try[T]
+
+ /**
+ * validate config param
+ *
+ * @param param param to be validated
+ * @return param itself
+ */
+ protected def validate[T <: Param](param: T): T = {
+ param.validate()
+ param
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala
new file mode 100644
index 0000000..f48d328
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala
@@ -0,0 +1,48 @@
+/*
+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.configuration.dqdefinition.reader
+
+import org.apache.griffin.measure.utils.JsonUtil
+
+object ParamReaderFactory {
+
+ val json = "json"
+ val file = "file"
+
+ /**
+ * parse string content to get param reader
+ * @param pathOrJson
+ * @return
+ */
+ def getParamReader(pathOrJson: String): ParamReader = {
+ val strType = paramStrType(pathOrJson)
+ if (json.equals(strType)) ParamJsonReader(pathOrJson)
+ else ParamFileReader(pathOrJson)
+ }
+
+ private def paramStrType(str: String): String = {
+ try {
+ JsonUtil.toAnyMap(str)
+ json
+ } catch {
+ case e: Throwable => file
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/DQConfig.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/DQConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/DQConfig.scala
deleted file mode 100644
index d13f61e..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/DQConfig.scala
+++ /dev/null
@@ -1,217 +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.configuration.params
-
-import com.fasterxml.jackson.annotation.JsonInclude.Include
-import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty}
-import org.apache.commons.lang.StringUtils
-import org.apache.griffin.measure.configuration.enums._
-
-/**
- * dq param
- * @param name name of dq measurement (must)
- * @param timestamp default timestamp of measure in batch mode (optional)
- * @param procType batch mode or streaming mode (must)
- * @param dataSources data sources (must)
- * @param evaluateRule dq measurement (must)
- */
-@JsonInclude(Include.NON_NULL)
-case class DQConfig(@JsonProperty("name") name: String,
- @JsonProperty("timestamp") timestamp: Long,
- @JsonProperty("process.type") procType: String,
- @JsonProperty("data.sources") dataSources: List[DataSourceParam],
- @JsonProperty("evaluate.rule") evaluateRule: EvaluateRuleParam
- ) extends Param {
- def getName: String = name
- def getTimestampOpt: Option[Long] = if (timestamp != null) Some(timestamp) else None
- def getProcType: String = procType
- def getDataSources: Seq[DataSourceParam] = {
- dataSources.foldLeft((Nil: Seq[DataSourceParam], Set[String]())) { (ret, ds) =>
- val (seq, names) = ret
- if (!names.contains(ds.getName)){
- (seq :+ ds, names + ds.getName)
- } else ret
- }._1
- }
- def getEvaluateRule: EvaluateRuleParam = evaluateRule
-
- def validate(): Unit = {
- assert(StringUtils.isNotBlank(name), "dq config name should not be blank")
- assert(StringUtils.isNotBlank(procType), "process.type should not be blank")
- assert((dataSources != null), "data.sources should not be null")
- assert((evaluateRule != null), "evaluate.rule should not be null")
- getDataSources.foreach(_.validate)
- evaluateRule.validate
- }
-}
-
-/**
- * data source param
- * @param name data source name (must)
- * @param connectors data connectors (optional)
- * @param cache data source cache configuration (must in streaming mode with streaming connectors)
- */
-@JsonInclude(Include.NON_NULL)
-case class DataSourceParam( @JsonProperty("name") name: String,
- @JsonProperty("connectors") connectors: List[DataConnectorParam],
- @JsonProperty("cache") cache: Map[String, Any]
- ) extends Param {
- def getName: String = name
- def getConnectors: Seq[DataConnectorParam] = if (connectors != null) connectors else Nil
- def getCacheOpt: Option[Map[String, Any]] = if (cache != null) Some(cache) else None
-
- def validate(): Unit = {
- assert(StringUtils.isNotBlank(name), "data source name should not be empty")
- getConnectors.foreach(_.validate)
- }
-}
-
-/**
- * data connector param
- * @param conType data connector type, e.g.: hive, avro, kafka (must)
- * @param version data connector type version (optional)
- * @param config detail configuration of data connector (must)
- * @param preProc pre-process rules after load data (optional)
- */
-@JsonInclude(Include.NON_NULL)
-case class DataConnectorParam( @JsonProperty("type") conType: String,
- @JsonProperty("version") version: String,
- @JsonProperty("config") config: Map[String, Any],
- @JsonProperty("pre.proc") preProc: List[RuleParam]
- ) extends Param {
- def getType: String = conType
- def getVersion: String = version
- def getConfig: Map[String, Any] = if (config != null) config else Map[String, Any]()
- def getPreProcRules: Seq[RuleParam] = if (preProc != null) preProc else Nil
-
- def validate(): Unit = {
- assert(StringUtils.isNotBlank(conType), "data connector type should not be empty")
- getPreProcRules.foreach(_.validate)
- }
-}
-
-/**
- * evaluate rule param
- * @param rules rules to define dq measurement (optional)
- */
-@JsonInclude(Include.NON_NULL)
-case class EvaluateRuleParam( @JsonProperty("rules") rules: List[RuleParam]
- ) extends Param {
- def getRules: Seq[RuleParam] = if (rules != null) rules else Nil
-
- def validate(): Unit = {
- getRules.foreach(_.validate)
- }
-}
-
-/**
- * rule param
- * @param dslType dsl type of this rule (must)
- * @param dqType dq type of this rule (must if dsl type is "griffin-dsl")
- * @param name name of result calculated by this rule (must if for later usage)
- * @param rule rule to define dq step calculation (must)
- * @param details detail config of rule (optional)
- * @param cache cache the result for multiple usage (optional, valid for "spark-sql" and "df-opr" mode)
- * @param metric config for metric output (optional)
- * @param record config for record output (optional)
- * @param dsCacheUpdate config for data source cache update output (optional, valid in streaming mode)
- */
-@JsonInclude(Include.NON_NULL)
-case class RuleParam( @JsonProperty("dsl.type") dslType: String,
- @JsonProperty("dq.type") dqType: String,
- @JsonProperty("name") name: String,
- @JsonProperty("rule") rule: String,
- @JsonProperty("details") details: Map[String, Any],
- @JsonProperty("cache") cache: Boolean,
- @JsonProperty("metric") metric: RuleMetricParam,
- @JsonProperty("record") record: RuleRecordParam,
- @JsonProperty("ds.cache.update") dsCacheUpdate: RuleDsCacheUpdateParam
- ) extends Param {
- def getDslType: DslType = if (dslType != null) DslType(dslType) else DslType("")
- def getDqType: DqType = if (dqType != null) DqType(dqType) else DqType("")
- def getCache: Boolean = if (cache != null) cache else false
-
- def getName: String = if (name != null) name else ""
- def getRule: String = if (rule != null) rule else ""
- def getDetails: Map[String, Any] = if (details != null) details else Map[String, Any]()
-
- def getMetricOpt: Option[RuleMetricParam] = if (metric != null) Some(metric) else None
- def getRecordOpt: Option[RuleRecordParam] = if (record != null) Some(record) else None
- def getDsCacheUpdateOpt: Option[RuleDsCacheUpdateParam] = if (dsCacheUpdate != null) Some(dsCacheUpdate) else None
-
- def replaceName(newName: String): RuleParam = {
- if (StringUtils.equals(newName, name)) this
- else RuleParam(dslType, dqType, newName, rule, details, cache, metric, record, dsCacheUpdate)
- }
- def replaceRule(newRule: String): RuleParam = {
- if (StringUtils.equals(newRule, rule)) this
- else RuleParam(dslType, dqType, name, newRule, details, cache, metric, record, dsCacheUpdate)
- }
- def replaceDetails(newDetails: Map[String, Any]): RuleParam = {
- RuleParam(dslType, dqType, name, rule, newDetails, cache, metric, record, dsCacheUpdate)
- }
-
- def validate(): Unit = {
- assert(!(getDslType.equals(GriffinDslType) && getDqType.equals(UnknownType)),
- "unknown dq type for griffin dsl")
-
- getMetricOpt.foreach(_.validate)
- getRecordOpt.foreach(_.validate)
- getDsCacheUpdateOpt.foreach(_.validate)
- }
-}
-
-/**
- * metric param of rule
- * @param name name of metric to output (optional)
- * @param collectType the normalize strategy to collect metric (optional)
- */
-@JsonInclude(Include.NON_NULL)
-case class RuleMetricParam( @JsonProperty("name") name: String,
- @JsonProperty("collect.type") collectType: String
- ) extends Param {
- def getNameOpt: Option[String] = if (StringUtils.isNotBlank(name)) Some(name) else None
- def getCollectType: NormalizeType = if (StringUtils.isNotBlank(collectType)) NormalizeType(collectType) else NormalizeType("")
-
- def validate(): Unit = {}
-}
-
-/**
- * record param of rule
- * @param name name of record to output (optional)
- */
-@JsonInclude(Include.NON_NULL)
-case class RuleRecordParam( @JsonProperty("name") name: String
- ) extends Param {
- def getNameOpt: Option[String] = if (StringUtils.isNotBlank(name)) Some(name) else None
-
- def validate(): Unit = {}
-}
-
-/**
- * data source cache update param of rule
- * @param dsName name of data source to be updated by thie rule result (must)
- */
-@JsonInclude(Include.NON_NULL)
-case class RuleDsCacheUpdateParam( @JsonProperty("ds.name") dsName: String
- ) extends Param {
- def getDsNameOpt: Option[String] = if (StringUtils.isNotBlank(dsName)) Some(dsName) else None
-
- def validate(): Unit = {}
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/EnvConfig.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/EnvConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/EnvConfig.scala
deleted file mode 100644
index fc4b984..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/EnvConfig.scala
+++ /dev/null
@@ -1,112 +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.configuration.params
-
-import com.fasterxml.jackson.annotation.JsonInclude.Include
-import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty}
-import org.apache.commons.lang.StringUtils
-import org.apache.griffin.measure.utils.TimeUtil
-
-/**
- * environment param
- * @param sparkParam config of spark environment (must)
- * @param persistParams config of persist ways (optional)
- * @param offsetCacheParams config of information cache ways (required in streaming mode)
- */
-@JsonInclude(Include.NON_NULL)
-case class EnvConfig(@JsonProperty("spark") sparkParam: SparkParam,
- @JsonProperty("persist") persistParams: List[PersistParam],
- @JsonProperty("info.cache") offsetCacheParams: List[OffsetCacheParam]
- ) extends Param {
- def getSparkParam: SparkParam = sparkParam
- def getPersistParams: Seq[PersistParam] = if (persistParams != null) persistParams else Nil
- def getOffsetCacheParams: Seq[OffsetCacheParam] = if (offsetCacheParams != null) offsetCacheParams else Nil
-
- def validate(): Unit = {
- assert((sparkParam != null), "spark param should not be null")
- sparkParam.validate
- getPersistParams.foreach(_.validate)
- getOffsetCacheParams.foreach(_.validate)
- }
-}
-
-/**
- * spark param
- * @param logLevel log level of spark application (optional)
- * @param cpDir checkpoint directory for spark streaming (required in streaming mode)
- * @param batchInterval batch interval for spark streaming (required in streaming mode)
- * @param processInterval process interval for streaming dq calculation (required in streaming mode)
- * @param config extra config for spark environment (optional)
- * @param initClear clear checkpoint directory or not when initial (optional)
- */
-@JsonInclude(Include.NON_NULL)
-case class SparkParam( @JsonProperty("log.level") logLevel: String,
- @JsonProperty("checkpoint.dir") cpDir: String,
- @JsonProperty("batch.interval") batchInterval: String,
- @JsonProperty("process.interval") processInterval: String,
- @JsonProperty("config") config: Map[String, String],
- @JsonProperty("init.clear") initClear: Boolean
- ) extends Param {
- def getLogLevel: String = if (logLevel != null) logLevel else "WARN"
- def getCpDir: String = if (cpDir != null) cpDir else ""
- def getBatchInterval: String = if (batchInterval != null) batchInterval else ""
- def getProcessInterval: String = if (processInterval != null) processInterval else ""
- def getConfig: Map[String, String] = if (config != null) config else Map[String, String]()
- def needInitClear: Boolean = if (initClear != null) initClear else false
-
- def validate(): Unit = {
-// assert(StringUtils.isNotBlank(cpDir), "checkpoint.dir should not be empty")
-// assert(TimeUtil.milliseconds(getBatchInterval).nonEmpty, "batch.interval should be valid time string")
-// assert(TimeUtil.milliseconds(getProcessInterval).nonEmpty, "process.interval should be valid time string")
- }
-}
-
-/**
- * persist param
- * @param persistType persist type, e.g.: log, hdfs, http, mongo (must)
- * @param config config of persist way (must)
- */
-@JsonInclude(Include.NON_NULL)
-case class PersistParam( @JsonProperty("type") persistType: String,
- @JsonProperty("config") config: Map[String, Any]
- ) extends Param {
- def getType: String = persistType
- def getConfig: Map[String, Any] = if (config != null) config else Map[String, Any]()
-
- def validate(): Unit = {
- assert(StringUtils.isNotBlank(persistType), "persist type should not be empty")
- }
-}
-
-/**
- * offset cache param
- * @param cacheType offset cache type, e.g.: zookeeper (must)
- * @param config config of cache way
- */
-@JsonInclude(Include.NON_NULL)
-case class OffsetCacheParam(@JsonProperty("type") cacheType: String,
- @JsonProperty("config") config: Map[String, Any]
- ) extends Param {
- def getType: String = cacheType
- def getConfig: Map[String, Any] = if (config != null) config else Map[String, Any]()
-
- def validate(): Unit = {
- assert(StringUtils.isNotBlank(cacheType), "info cache type should not be empty")
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/GriffinConfig.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/GriffinConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/GriffinConfig.scala
deleted file mode 100644
index 8debe48..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/GriffinConfig.scala
+++ /dev/null
@@ -1,42 +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.configuration.params
-
-import com.fasterxml.jackson.annotation.{JsonInclude, JsonProperty}
-import com.fasterxml.jackson.annotation.JsonInclude.Include
-
-/**
- * full set of griffin configuration
- * @param envConfig environment configuration (must)
- * @param dqConfig dq measurement configuration (must)
- */
-@JsonInclude(Include.NON_NULL)
-case class GriffinConfig(@JsonProperty("env") envConfig: EnvConfig,
- @JsonProperty("dq") dqConfig: DQConfig
- ) extends Param {
- def getEnvConfig: EnvConfig = envConfig
- def getDqConfig: DQConfig = dqConfig
-
- def validate(): Unit = {
- assert((envConfig != null), "environment config should not be null")
- assert((dqConfig != null), "dq config should not be null")
- envConfig.validate
- dqConfig.validate
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/Param.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/Param.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/Param.scala
deleted file mode 100644
index 6116cdf..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/Param.scala
+++ /dev/null
@@ -1,28 +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.configuration.params
-
-trait Param extends Serializable {
-
- /**
- * validate param internally
- */
- def validate(): Unit
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamFileReader.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamFileReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamFileReader.scala
deleted file mode 100644
index a528127..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamFileReader.scala
+++ /dev/null
@@ -1,43 +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.configuration.params.reader
-
-import org.apache.griffin.measure.configuration.params.Param
-import org.apache.griffin.measure.utils.{HdfsUtil, JsonUtil}
-
-import scala.util.Try
-
-/**
- * read params from config file path
- * @param filePath: hdfs path ("hdfs://cluster-name/path")
- * local file path ("file:///path")
- * relative file path ("relative/path")
- */
-case class ParamFileReader(filePath: String) extends ParamReader {
-
- def readConfig[T <: Param](implicit m : Manifest[T]): Try[T] = {
- Try {
- val source = HdfsUtil.openFile(filePath)
- val param = JsonUtil.fromJson[T](source)
- source.close
- validate(param)
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamJsonReader.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamJsonReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamJsonReader.scala
deleted file mode 100644
index 91ffa9a..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamJsonReader.scala
+++ /dev/null
@@ -1,39 +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.configuration.params.reader
-
-import org.apache.griffin.measure.configuration.params.Param
-import org.apache.griffin.measure.utils.JsonUtil
-
-import scala.util.Try
-
-/**
- * read params from json string directly
- * @param jsonString
- */
-case class ParamJsonReader(jsonString: String) extends ParamReader {
-
- def readConfig[T <: Param](implicit m : Manifest[T]): Try[T] = {
- Try {
- val param = JsonUtil.fromJson[T](jsonString)
- validate(param)
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReader.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReader.scala
deleted file mode 100644
index 21ebccd..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReader.scala
+++ /dev/null
@@ -1,45 +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.configuration.params.reader
-
-import org.apache.griffin.measure.Loggable
-import org.apache.griffin.measure.configuration.params.Param
-
-import scala.util.Try
-
-trait ParamReader extends Loggable with Serializable {
-
- /**
- * read config param
- * @tparam T param type expected
- * @return parsed param
- */
- def readConfig[T <: Param](implicit m : Manifest[T]): Try[T]
-
- /**
- * validate config param
- * @param param param to be validated
- * @return param itself
- */
- protected def validate[T <: Param](param: T): T = {
- param.validate()
- param
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReaderFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReaderFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReaderFactory.scala
deleted file mode 100644
index 363b865..0000000
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/params/reader/ParamReaderFactory.scala
+++ /dev/null
@@ -1,48 +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.configuration.params.reader
-
-import org.apache.griffin.measure.utils.JsonUtil
-
-object ParamReaderFactory {
-
- val json = "json"
- val file = "file"
-
- /**
- * parse string content to get param reader
- * @param pathOrJson
- * @return
- */
- def getParamReader(pathOrJson: String): ParamReader = {
- val strType = paramStrType(pathOrJson)
- if (json.equals(strType)) ParamJsonReader(pathOrJson)
- else ParamFileReader(pathOrJson)
- }
-
- private def paramStrType(str: String): String = {
- try {
- JsonUtil.toAnyMap(str)
- json
- } catch {
- case e: Throwable => file
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 2956c9b..eee8917 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
@@ -19,7 +19,7 @@ under the License.
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.configuration.dqdefinition._
import org.apache.griffin.measure.datasource._
import org.apache.griffin.measure.sink.{Sink, SinkFactory}
import org.apache.spark.sql.{Encoders, SQLContext, SparkSession}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheClient.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheClient.scala b/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheClient.scala
index 416ff31..8d4abea 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheClient.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheClient.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.context.streaming.offset
-import org.apache.griffin.measure.configuration.params.OffsetCacheParam
+import org.apache.griffin.measure.configuration.dqdefinition.OffsetCacheParam
import org.apache.griffin.measure.context.streaming.lock.{CacheLock, CacheLockSeq}
object OffsetCacheClient extends OffsetCache with OffsetOps {
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheFactory.scala
index 07bca27..0fbf335 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheFactory.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/context/streaming/offset/OffsetCacheFactory.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.context.streaming.offset
-import org.apache.griffin.measure.configuration.params.OffsetCacheParam
+import org.apache.griffin.measure.configuration.dqdefinition.OffsetCacheParam
import scala.util.{Success, Try}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSource.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSource.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSource.scala
index 8b068e3..8bb0735 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSource.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSource.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.datasource
import org.apache.griffin.measure.Loggable
-import org.apache.griffin.measure.configuration.params.DataSourceParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataSourceParam
import org.apache.griffin.measure.datasource.cache.StreamingCacheClient
import org.apache.griffin.measure.context.{DQContext, TimeRange}
import org.apache.griffin.measure.datasource.connector.DataConnector
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSourceFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSourceFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSourceFactory.scala
index 888d92a..f48185e 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSourceFactory.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/DataSourceFactory.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.datasource
import org.apache.griffin.measure.Loggable
-import org.apache.griffin.measure.configuration.params.DataSourceParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataSourceParam
import org.apache.griffin.measure.datasource.cache.StreamingCacheClientFactory
import org.apache.griffin.measure.datasource.connector.{DataConnector, DataConnectorFactory}
import org.apache.spark.sql.SparkSession
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 1abbd30..fc09661 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
@@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicLong
import org.apache.griffin.measure.Loggable
import org.apache.griffin.measure.configuration.enums.{BatchProcessType, DslType, SparkSqlType}
-import org.apache.griffin.measure.configuration.params.DataConnectorParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.context.{ContextId, DQContext, TimeRange}
import org.apache.griffin.measure.datasource.TimestampStorage
import org.apache.griffin.measure.job.builder.DQJobBuilder
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
index 4edd4d2..104b05c 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.datasource.connector
import org.apache.griffin.measure.Loggable
-import org.apache.griffin.measure.configuration.params.DataConnectorParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.datasource.TimestampStorage
import org.apache.griffin.measure.datasource.cache.StreamingCacheClient
import org.apache.griffin.measure.datasource.connector.batch._
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 f5439fc..7fa9080 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
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.datasource.connector.batch
-import org.apache.griffin.measure.configuration.params.DataConnectorParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.context.TimeRange
import org.apache.griffin.measure.datasource.TimestampStorage
import org.apache.griffin.measure.utils.HdfsUtil
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 f6533b5..ab1e823 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
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.datasource.connector.batch
-import org.apache.griffin.measure.configuration.params.DataConnectorParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.context.TimeRange
import org.apache.griffin.measure.datasource.TimestampStorage
import org.apache.spark.sql.{DataFrame, SparkSession}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 fd83c1e..3dcab16 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
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.datasource.connector.batch
-import org.apache.griffin.measure.configuration.params.DataConnectorParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.context.TimeRange
import org.apache.griffin.measure.datasource.TimestampStorage
import org.apache.griffin.measure.utils.HdfsUtil
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingStringDataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingStringDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingStringDataConnector.scala
index c24aadc..9c7b0e2 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingStringDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingStringDataConnector.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.datasource.connector.streaming
import kafka.serializer.StringDecoder
-import org.apache.griffin.measure.configuration.params.DataConnectorParam
+import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.datasource.TimestampStorage
import org.apache.griffin.measure.datasource.cache.StreamingCacheClient
import org.apache.spark.rdd.RDD
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/job/builder/DQJobBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/job/builder/DQJobBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/job/builder/DQJobBuilder.scala
index 074a74e..249d6d2 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/job/builder/DQJobBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/job/builder/DQJobBuilder.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.job.builder
import org.apache.griffin.measure.configuration.enums.DslType
-import org.apache.griffin.measure.configuration.params._
+import org.apache.griffin.measure.configuration.dqdefinition._
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.job._
import org.apache.griffin.measure.step.DQStep
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 29cee08..8860408 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
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.launch
import org.apache.griffin.measure.Loggable
-import org.apache.griffin.measure.configuration.params.{DQConfig, EnvConfig}
+import org.apache.griffin.measure.configuration.dqdefinition.{DQConfig, EnvConfig}
import scala.util.Try
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 1174dbb..a651f2e 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
@@ -21,7 +21,7 @@ package org.apache.griffin.measure.launch.batch
import java.util.Date
import org.apache.griffin.measure.configuration.enums._
-import org.apache.griffin.measure.configuration.params._
+import org.apache.griffin.measure.configuration.dqdefinition._
import org.apache.griffin.measure.context._
import org.apache.griffin.measure.datasource.DataSourceFactory
import org.apache.griffin.measure.job.builder.DQJobBuilder
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 73fc7a0..4bd8c54 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
@@ -23,7 +23,7 @@ import java.util.concurrent.{Executors, ThreadPoolExecutor, TimeUnit}
import org.apache.griffin.measure.Loggable
import org.apache.griffin.measure.configuration.enums._
-import org.apache.griffin.measure.configuration.params._
+import org.apache.griffin.measure.configuration.dqdefinition._
import org.apache.griffin.measure.context._
import org.apache.griffin.measure.datasource.DataSourceFactory
import org.apache.griffin.measure.context.streaming.offset.OffsetCacheClient
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/sink/SinkFactory.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/SinkFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/SinkFactory.scala
index e909308..85bf1b1 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/SinkFactory.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/SinkFactory.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.sink
-import org.apache.griffin.measure.configuration.params.PersistParam
+import org.apache.griffin.measure.configuration.dqdefinition.PersistParam
import scala.util.{Success, Try}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/BatchDataSourceStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/BatchDataSourceStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/BatchDataSourceStepBuilder.scala
index 2d739c4..9ef383c 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/BatchDataSourceStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/BatchDataSourceStepBuilder.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.step.builder
-import org.apache.griffin.measure.configuration.params._
+import org.apache.griffin.measure.configuration.dqdefinition._
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.read.ReadStep
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/DQStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/DQStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/DQStepBuilder.scala
index 2ee308d..c37b5a3 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/DQStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/DQStepBuilder.scala
@@ -21,7 +21,7 @@ package org.apache.griffin.measure.step.builder
import org.apache.commons.lang.StringUtils
import org.apache.griffin.measure.Loggable
import org.apache.griffin.measure.configuration.enums._
-import org.apache.griffin.measure.configuration.params.{DataSourceParam, Param, RuleParam}
+import org.apache.griffin.measure.configuration.dqdefinition.{DataSourceParam, Param, RuleParam}
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step._
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataFrameOpsDQStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataFrameOpsDQStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataFrameOpsDQStepBuilder.scala
index c3854f9..f945856 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataFrameOpsDQStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataFrameOpsDQStepBuilder.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.step.builder
-import org.apache.griffin.measure.configuration.params.RuleParam
+import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.DQStep
import org.apache.griffin.measure.step.transform.DataFrameOpsTransformStep
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataSourceParamStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataSourceParamStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataSourceParamStepBuilder.scala
index 333615d..6d3922a 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataSourceParamStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/DataSourceParamStepBuilder.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.step.builder
-import org.apache.griffin.measure.configuration.params.{DataConnectorParam, DataSourceParam}
+import org.apache.griffin.measure.configuration.dqdefinition.{DataConnectorParam, DataSourceParam}
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.DQStep
import org.apache.griffin.measure.step.read._
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/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 7ce78d1..7dd3c73 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
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.step.builder
-import org.apache.griffin.measure.configuration.params.RuleParam
+import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.DQStep
import org.apache.griffin.measure.step.builder.dsl.parser.GriffinDslParser
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/RuleParamStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/RuleParamStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/RuleParamStepBuilder.scala
index 2a43b34..f9f101e 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/RuleParamStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/RuleParamStepBuilder.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.step.builder
import org.apache.griffin.measure.configuration.enums.NormalizeType
-import org.apache.griffin.measure.configuration.params.RuleParam
+import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.write.{DataSourceUpdateWriteStep, MetricWriteStep, RecordWriteStep}
import org.apache.griffin.measure.step.{DQStep, SeqDQStep}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/SparkSqlDQStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/SparkSqlDQStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/SparkSqlDQStepBuilder.scala
index 36237d7..f66192c 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/SparkSqlDQStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/SparkSqlDQStepBuilder.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.step.builder
-import org.apache.griffin.measure.configuration.params.RuleParam
+import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.DQStep
import org.apache.griffin.measure.step.transform.SparkSqlTransformStep
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/StreamingDataSourceStepBuilder.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/StreamingDataSourceStepBuilder.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/StreamingDataSourceStepBuilder.scala
index 0139d19..24ea36d 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/StreamingDataSourceStepBuilder.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/StreamingDataSourceStepBuilder.scala
@@ -18,7 +18,7 @@ under the License.
*/
package org.apache.griffin.measure.step.builder
-import org.apache.griffin.measure.configuration.params._
+import org.apache.griffin.measure.configuration.dqdefinition._
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.read.ReadStep
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/b9e0743f/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/AccuracyExpr2DQSteps.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/AccuracyExpr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/AccuracyExpr2DQSteps.scala
index 0416dbb..27872c2 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/AccuracyExpr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/AccuracyExpr2DQSteps.scala
@@ -19,7 +19,7 @@ under the License.
package org.apache.griffin.measure.step.builder.dsl.transform
import org.apache.griffin.measure.configuration.enums.{BatchProcessType, NormalizeType, StreamingProcessType}
-import org.apache.griffin.measure.configuration.params.RuleParam
+import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.DQStep
import org.apache.griffin.measure.step.builder.ConstantColumns