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 88bcc16..b57c9c5 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/Application.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/Application.scala
@@ -119,6 +119,8 @@
 
     if (!success) {
       sys.exit(-5)
+    } else {
+      sys.exit(0)
     }
   }
 
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 fa1468c..c72f06c 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
@@ -66,15 +66,15 @@
   implicit val encoder: Encoder[String] = Encoders.STRING
   val functionNames: Seq[String] = sparkSession.catalog.listFunctions.map(_.name).collect.toSeq
 
-  val dataSourceTimeRanges: Map[String, TimeRange] = loadDataSources()
+  var dataSourceTimeRanges: Map[String, TimeRange] = _
 
   def loadDataSources(): Map[String, TimeRange] = {
-    dataSources.map { ds =>
+    dataSourceTimeRanges = dataSources.map { ds =>
       (ds.name, ds.loadData(this))
     }.toMap
-  }
 
-  printTimeRanges()
+    dataSourceTimeRanges
+  }
 
   private val sinkFactory = SinkFactory(sinkParams, name)
   private val defaultSinks: Seq[Sink] = createSinks(contextId.timestamp)
@@ -105,16 +105,4 @@
     dataFrameCache.clearAllTrashDataFrames()
   }
 
-  private def printTimeRanges(): Unit = {
-    if (dataSourceTimeRanges.nonEmpty) {
-      val timeRangesStr = dataSourceTimeRanges
-        .map { pair =>
-          val (name, timeRange) = pair
-          s"$name -> (${timeRange.begin}, ${timeRange.end}]"
-        }
-        .mkString(", ")
-      println(s"data source timeRanges: $timeRangesStr")
-    }
-  }
-
 }
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala
index 0454f38..a087d1b 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala
@@ -27,23 +27,54 @@
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.utils.ParamUtil._
 
+/**
+ * Measure
+ *
+ * An abstraction for a data quality measure implementation.
+ */
 trait Measure extends Loggable {
   import Measure._
 
+  /**
+   * SparkSession for this Griffin Application.
+   */
   val sparkSession: SparkSession
 
+  /**
+   * Object representation of user defined measure.
+   */
   val measureParam: MeasureParam
 
+  /**
+   * If this measure supports record writing.
+   */
   val supportsRecordWrite: Boolean
 
+  /**
+   * If this measure supports metric writing.
+   */
   val supportsMetricWrite: Boolean
 
+  /**
+   * Metric values column.
+   */
   final val valueColumn = s"${MeasureColPrefix}_${measureParam.getName}"
 
+  /**
+   * Helper method to get a typed value from measure configuration based on given key.
+   *
+   * @param key given key for which the value needs to be fetched.
+   * @param defValue default value in case of no value.
+   * @tparam T type of value to get.
+   * @return value for given key
+   */
   def getFromConfig[T: ClassTag](key: String, defValue: T): T = {
     measureParam.getConfig.getAnyRef[T](key, defValue)
   }
 
+  /**
+   * Enriches metrics dataframe with some additional keys.
+   */
   // todo add status col to persist blank metrics if the measure fails
   def preProcessMetrics(input: DataFrame): DataFrame = {
     if (supportsMetricWrite) {
@@ -56,6 +87,9 @@
     } else input
   }
 
+  /**
+   * Enriches records dataframe with a status column marking rows as good or bad based on values.
+   */
   def preProcessRecords(input: DataFrame): DataFrame = {
     if (supportsRecordWrite) {
       input
@@ -64,38 +98,59 @@
     } else input
   }
 
+  /**
+   * Implementation of this measure.
+   *
+   * @return tuple of records dataframe and metric dataframe
+   */
   def impl(): (DataFrame, DataFrame)
 
+  /**
+   * Implementation should define validtion checks in this method (if required).
+   * This method needs to be called explicitly call this method (preferably during measure creation).
+   *
+   * Defaults to no-op.
+   */
   def validate(): Unit = {}
 
+  /**
+   * Executes this measure specific transformation on input data source.
+   *
+   * @param batchId batch id to append in case of streaming source.
+   * @return enriched tuple of records dataframe and metric dataframe
+   */
   def execute(batchId: Option[Long] = None): (DataFrame, DataFrame) = {
     val (recordsDf, metricDf) = impl()
 
     val processedRecordDf = preProcessRecords(recordsDf)
     val processedMetricDf = preProcessMetrics(metricDf)
 
-    var batchDetailsOpt = StringUtils.EMPTY
     val res = batchId match {
       case Some(batchId) =>
         implicit val bId: Long = batchId
-        batchDetailsOpt = s"for batch id $bId"
         (appendBatchIdIfAvailable(processedRecordDf), appendBatchIdIfAvailable(processedMetricDf))
       case None => (processedRecordDf, processedMetricDf)
     }
 
-    info(
-      s"Execution of '${measureParam.getType}' measure " +
-        s"with name '${measureParam.getName}' is complete $batchDetailsOpt")
-
     res
   }
 
+  /**
+   * Appends batch id to metrics in case of streaming sources.
+   *
+   * @param input metric dataframe
+   * @param batchId batch id to append
+   * @return updated metric dataframe
+   */
   private def appendBatchIdIfAvailable(input: DataFrame)(implicit batchId: Long): DataFrame = {
     input.withColumn(BatchId, typedLit[Long](batchId))
   }
 
 }
 
+/**
+ * Measure Constants.
+ */
 object Measure {
 
   final val DataSource = "data_source"
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala
index 1b75f2f..1f49328 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala
@@ -17,19 +17,79 @@
 
 package org.apache.griffin.measure.execution
 
-import org.apache.spark.sql.DataFrame
+import java.util.Date
+import java.util.concurrent.Executors
+
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService, Future}
+import scala.util.{Failure, Success}
+
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.sql.{DataFrame, SparkSession}
 
 import org.apache.griffin.measure.Loggable
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.configuration.enums.{MeasureTypes, OutputType}
-import org.apache.griffin.measure.context.DQContext
+import org.apache.griffin.measure.context.{ContextId, DQContext}
 import org.apache.griffin.measure.execution.impl._
 import org.apache.griffin.measure.step.write.{MetricFlushStep, MetricWriteStep, RecordWriteStep}
 
+/**
+ * MeasureExecutor
+ *
+ * This acts as the starting point for the execution of different data quality measures
+ * defined by the users in `DQConfig`. Execution of the measures involves the following steps,
+ *  - Create a fix pool of threads which will be used to execute measures in parallel
+ *  - For each measure defined per data source,
+ *    - Caching data source(s) if necessary
+ *    - In parallel do the following,
+ *      - Create Measure entity (transformation step)
+ *      - Write Metrics if required and if supported (metric write step)
+ *      - Write Records if required and if supported (record write step)
+ *      - Clear internal objects (metric flush step)
+ *    - Un caching data source(s) if cached already.
+ *
+ * In contrast to the execution of `GriffinDslDQStepBuilder`, `MeasureExecutor` executes each of
+ * the defined measures independently. This means that the outputs (metrics and records) are written
+ * independently for each measure.
+ *
+ * @param context Instance of `DQContext`
+ */
 case class MeasureExecutor(context: DQContext) extends Loggable {
 
+  /**
+   * SparkSession for this Griffin Application.
+   */
+  private val sparkSession: SparkSession = context.sparkSession
+
+  /**
+   * Enable or disable caching of data sources before execution. Defaults to `true`.
+   */
+  private val cacheDataSources: Boolean = sparkSession.sparkContext.getConf
+    .getBoolean("spark.griffin.measure.cacheDataSources", defaultValue = true)
+
+  /**
+   * Size of thread pool for parallel measure execution.
+   * Defaults to number of processors available to the spark driver JVM.
+   */
+  private val numThreads: Int = sparkSession.sparkContext.getConf
+    .getInt("spark.griffin.measure.parallelism", Runtime.getRuntime.availableProcessors())
+
+  /**
+   * Service to handle threaded execution of tasks (measures).
+   */
+  private implicit val ec: ExecutionContextExecutorService =
+    ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
+
+  /**
+   * Starting point of measure execution.
+   *
+   * @param measureParams Object representation(s) of user defined measure(s).
+   */
   def execute(measureParams: Seq[MeasureParam]): Unit = {
-    cacheIfNecessary(measureParams)
+    implicit val measureCountByDataSource: Map[String, Int] = measureParams
+      .map(_.getDataSource)
+      .groupBy(x => x)
+      .mapValues(_.length)
 
     measureParams
       .groupBy(measureParam => measureParam.getDataSource)
@@ -37,50 +97,114 @@
         val dataSourceName = measuresForSource._1
         val measureParams = measuresForSource._2
 
-        val dataSource = context.sparkSession.read.table(dataSourceName)
+        withCacheIfNecessary(dataSourceName, {
+          val dataSource = sparkSession.read.table(dataSourceName)
 
-        if (dataSource.isStreaming) {
-          // todo this is a no op as streaming queries need to be registered.
-          dataSource.writeStream
-            .foreachBatch((_, batchId) => {
-              executeMeasures(measureParams, Some(batchId))
-            })
-        } else {
-          executeMeasures(measureParams)
-        }
+          if (dataSource.isStreaming) {
+            // TODO this is a no op as streaming queries need to be registered.
+
+            dataSource.writeStream
+              .foreachBatch((_, batchId) => {
+                executeMeasures(measureParams, Some(batchId))
+              })
+          } else {
+            executeMeasures(measureParams)
+          }
+        })
       })
   }
 
-  private def cacheIfNecessary(measureParams: Seq[MeasureParam]): Unit = {
-    measureParams
-      .map(_.getDataSource)
-      .groupBy(x => x)
-      .mapValues(_.length)
-      .filter(_._2 > 1)
-      .foreach(source => {
-        info(
-          s"Caching data source with name '${source._1}'" +
-            s" as ${source._2} measures are applied on it.")
-        context.sparkSession.catalog.cacheTable(source._1)
-      })
+  /**
+   * Performs a function with cached data source if necessary.
+   * Caches data sources if it has more than 1 measure defined for it, and, `cacheDataSources` is `true`.
+   * After the function is complete, the data source is uncached.
+   *
+   * @param dataSourceName name of data source
+   * @param f function to perform
+   * @param measureCountByDataSource number of measures for each data source
+   * @tparam T return type of function `f`
+   * @return
+   */
+  private def withCacheIfNecessary[T](dataSourceName: String, f: => T)(
+      implicit measureCountByDataSource: Map[String, Int]): T = {
+    val numMeasures = measureCountByDataSource(dataSourceName)
+    var isCached = false
+    if (cacheDataSources && numMeasures > 1) {
+      info(
+        s"Caching data source with name '$dataSourceName' as $numMeasures measures are applied on it.")
+      sparkSession.catalog.cacheTable(dataSourceName)
+      isCached = true
+    }
+
+    f
+
+    if (isCached) {
+      sparkSession.catalog.uncacheTable(dataSourceName)
+    }
   }
 
+  /**
+   * Executes measures for a data sources. Involves the following steps,
+   *  - Transformation
+   *  - Persist metrics if required
+   *  - Persist records if required
+   *
+   *  All measures are executed in parallel.
+   *
+   * @param measureParams Object representation(s) of user defined measure(s).
+   * @param batchId Option batch Id in case of streaming sources to identify micro batches.
+   */
   private def executeMeasures(
       measureParams: Seq[MeasureParam],
       batchId: Option[Long] = None): Unit = {
-    measureParams.foreach(measureParam => {
-      val measure = createMeasure(measureParam)
-      val (recordsDf, metricsDf) = measure.execute(batchId)
+    val batchDetailsOpt = batchId.map(bId => s"for batch id $bId").getOrElse(StringUtils.EMPTY)
 
-      persistRecords(measure, recordsDf)
-      persistMetrics(measure, metricsDf)
+    // define the tasks
+    val tasks: Map[String, Future[_]] = (for (i <- measureParams.indices)
+      yield {
+        val measureParam = measureParams(i)
+        val measureName = measureParam.getName
 
-      MetricFlushStep().execute(context)
+        (measureName, Future {
+          val currentContext = context.cloneDQContext(ContextId(new Date().getTime))
+          info(s"Started execution of measure with name '$measureName'")
+
+          val measure = createMeasure(measureParam)
+          val (recordsDf, metricsDf) = measure.execute(batchId)
+
+          persistRecords(currentContext, measure, recordsDf)
+          persistMetrics(currentContext, measure, metricsDf)
+
+          MetricFlushStep().execute(currentContext)
+        })
+      }).toMap
+
+    tasks.foreach(task =>
+      task._2.onComplete {
+        case Success(_) =>
+          info(s"Successfully executed measure with name '${task._1}' $batchDetailsOpt")
+        case Failure(e) =>
+          error(s"Error executing measure with name '${task._1}' $batchDetailsOpt", e)
     })
+
+    while (!tasks.forall(_._2.isCompleted)) {
+      info(
+        s"Measures with name ${tasks.filterNot(_._2.isCompleted).keys.mkString("['", "', '", "']")} " +
+          s"are still executing.")
+      Thread.sleep(1000)
+    }
+
+    info(
+      s"Completed execution of all measures for data source with name '${measureParams.head.getDataSource}'.")
   }
 
+  /**
+   * Instantiates measure implementations based on the user defined configurations.
+   *
+   * @param measureParam Object representation of user defined a measure.
+   * @return
+   */
   private def createMeasure(measureParam: MeasureParam): Measure = {
-    val sparkSession = context.sparkSession
     measureParam.getType match {
       case MeasureTypes.Completeness => CompletenessMeasure(sparkSession, measureParam)
       case MeasureTypes.Duplication => DuplicationMeasure(sparkSession, measureParam)
@@ -95,7 +219,14 @@
     }
   }
 
-  private def persistRecords(measure: Measure, recordsDf: DataFrame): Unit = {
+  /**
+   * Persists records to one or more sink based on the user defined measure configuration.
+   *
+   * @param context DQ Context.
+   * @param measure a measure implementation
+   * @param recordsDf records dataframe to persist.
+   */
+  private def persistRecords(context: DQContext, measure: Measure, recordsDf: DataFrame): Unit = {
     val measureParam: MeasureParam = measure.measureParam
 
     measureParam.getOutputOpt(OutputType.RecordOutputType) match {
@@ -108,7 +239,14 @@
     }
   }
 
-  private def persistMetrics(measure: Measure, metricsDf: DataFrame): Unit = {
+  /**
+   * Persists metrics to one or more sink based on the user defined measure configuration.
+   *
+   * @param context DQ Context.
+   * @param measure a measure implementation
+   * @param metricsDf metrics dataframe to persist
+   */
+  private def persistMetrics(context: DQContext, measure: Measure, metricsDf: DataFrame): Unit = {
     val measureParam: MeasureParam = measure.measureParam
 
     measureParam.getOutputOpt(OutputType.MetricOutputType) match {
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 9c03f98..7e97dc7 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
@@ -77,6 +77,8 @@
           dqContext = DQContext(contextId, metricName, dataSources, sinkParams, BatchProcessType)(
             sparkSession)
 
+          dqContext.loadDataSources()
+
           // start id
           val applicationId = sparkSession.sparkContext.applicationId
           dqContext.getSinks.foreach(_.open(applicationId))
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 09554f3..c83acb3 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
@@ -98,6 +98,7 @@
     val globalContext: DQContext =
       DQContext(contextId, metricName, dataSources, sinkParams, StreamingProcessType)(
         sparkSession)
+    globalContext.loadDataSources()
 
     // start id
     val applicationId = sparkSession.sparkContext.applicationId