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 796c797..743b05d 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
@@ -29,7 +29,7 @@
     val name = getStepName(ruleParam.getOutDfName())
     val inputDfName = getStepName(ruleParam.getInDfName())
     val transformStep = DataFrameOpsTransformStep(
-      name, inputDfName, ruleParam.getRule, ruleParam.getDetails, ruleParam.getCache)
+      name, inputDfName, ruleParam.getRule, ruleParam.getDetails, None, ruleParam.getCache)
     transformStep +: buildDirectWriteSteps(ruleParam)
   }
 
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 b5dfd0c..0fdf20a 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
@@ -28,7 +28,7 @@
   def buildSteps(context: DQContext, ruleParam: RuleParam): Seq[DQStep] = {
     val name = getStepName(ruleParam.getOutDfName())
     val transformStep = SparkSqlTransformStep(
-      name, ruleParam.getRule, ruleParam.getDetails, ruleParam.getCache)
+      name, ruleParam.getRule, ruleParam.getDetails, None, ruleParam.getCache)
     transformStep +: buildDirectWriteSteps(ruleParam)
   }
 
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 31eef69..3bb5737 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
@@ -81,25 +81,22 @@
         s"SELECT ${selClause} FROM `${sourceName}` " +
           s"LEFT JOIN `${targetName}` ON ${onClause} WHERE ${whereClause}"
       }
-      val missRecordsTransStep =
-        SparkSqlTransformStep(missRecordsTableName, missRecordsSql, emptyMap, true)
 
       val missRecordsWriteSteps = procType match {
         case BatchProcessType =>
           val rwName =
             ruleParam.getOutputOpt(RecordOutputType).
               flatMap(_.getNameOpt).getOrElse(missRecordsTableName)
-          RecordWriteStep(rwName, missRecordsTableName) :: Nil
-        case StreamingProcessType => Nil
-      }
-      val missRecordsUpdateWriteSteps = procType match {
-        case BatchProcessType => Nil
+          RecordWriteStep(rwName, missRecordsTableName)
         case StreamingProcessType =>
           val dsName =
             ruleParam.getOutputOpt(DscUpdateOutputType).flatMap(_.getNameOpt).getOrElse(sourceName)
-          DataSourceUpdateWriteStep(dsName, missRecordsTableName) :: Nil
+          DataSourceUpdateWriteStep(dsName, missRecordsTableName)
       }
 
+      val missRecordsTransStep =
+        SparkSqlTransformStep(missRecordsTableName, missRecordsSql, emptyMap, Some(missRecordsWriteSteps), true)
+
       // 2. miss count
       val missCountTableName = "__missCount"
       val missColName = details.getStringOrKey(_miss)
@@ -151,23 +148,23 @@
              |ON `${totalCountTableName}`.`${ConstantColumns.tmst}` = `${missCountTableName}`.`${ConstantColumns.tmst}`
          """.stripMargin
       }
-      val accuracyTransStep = SparkSqlTransformStep(accuracyTableName, accuracyMetricSql, emptyMap)
-      accuracyTransStep.parentSteps += missCountTransStep
-      accuracyTransStep.parentSteps += totalCountTransStep
-      val accuracyMetricWriteSteps = procType match {
+
+      val accuracyMetricWriteStep = procType match {
         case BatchProcessType =>
           val metricOpt = ruleParam.getOutputOpt(MetricOutputType)
           val mwName = metricOpt.flatMap(_.getNameOpt).getOrElse(ruleParam.getOutDfName())
           val flattenType = metricOpt.map(_.getFlatten).getOrElse(FlattenType.default)
-          MetricWriteStep(mwName, accuracyTableName, flattenType) :: Nil
-        case StreamingProcessType => Nil
+          Some(MetricWriteStep(mwName, accuracyTableName, flattenType))
+        case StreamingProcessType => None
       }
 
-      val batchWriteSteps =
-        accuracyMetricWriteSteps ++ missRecordsWriteSteps ++ missRecordsUpdateWriteSteps
+      val accuracyTransStep =
+        SparkSqlTransformStep(accuracyTableName, accuracyMetricSql, emptyMap, accuracyMetricWriteStep)
+      accuracyTransStep.parentSteps += missCountTransStep
+      accuracyTransStep.parentSteps += totalCountTransStep
 
       procType match {
-        case BatchProcessType => accuracyTransStep :: batchWriteSteps
+        case BatchProcessType => accuracyTransStep :: Nil
         // streaming extra steps
         case StreamingProcessType =>
           // 5. accuracy metric merge
@@ -178,15 +175,16 @@
             (AccuracyOprKeys._total -> totalColName),
             (AccuracyOprKeys._matched -> matchedColName)
           )
-          val accuracyMetricTransStep = DataFrameOpsTransformStep(accuracyMetricTableName,
-            accuracyTableName, accuracyMetricRule, accuracyMetricDetails)
-          accuracyMetricTransStep.parentSteps += accuracyTransStep
           val accuracyMetricWriteStep = {
             val metricOpt = ruleParam.getOutputOpt(MetricOutputType)
             val mwName = metricOpt.flatMap(_.getNameOpt).getOrElse(ruleParam.getOutDfName())
             val flattenType = metricOpt.map(_.getFlatten).getOrElse(FlattenType.default)
             MetricWriteStep(mwName, accuracyMetricTableName, flattenType)
           }
+          val accuracyMetricTransStep = DataFrameOpsTransformStep(accuracyMetricTableName,
+            accuracyTableName, accuracyMetricRule, accuracyMetricDetails, Some(accuracyMetricWriteStep))
+          accuracyMetricTransStep.parentSteps += accuracyTransStep
+
 
           // 6. collect accuracy records
           val accuracyRecordTableName = "__accuracyRecords"
@@ -196,9 +194,7 @@
                |FROM `${accuracyMetricTableName}` WHERE `${ConstantColumns.record}`
              """.stripMargin
           }
-          val accuracyRecordTransStep = SparkSqlTransformStep(
-            accuracyRecordTableName, accuracyRecordSql, emptyMap)
-          accuracyRecordTransStep.parentSteps += accuracyMetricTransStep
+
           val accuracyRecordWriteStep = {
             val rwName =
               ruleParam.getOutputOpt(RecordOutputType).flatMap(_.getNameOpt)
@@ -206,10 +202,11 @@
 
             RecordWriteStep(rwName, missRecordsTableName, Some(accuracyRecordTableName))
           }
+          val accuracyRecordTransStep = SparkSqlTransformStep(
+            accuracyRecordTableName, accuracyRecordSql, emptyMap, Some(accuracyRecordWriteStep))
+          accuracyRecordTransStep.parentSteps += accuracyMetricTransStep
 
-          // extra steps
-          val streamingWriteSteps = accuracyMetricWriteStep :: accuracyRecordWriteStep :: Nil
-          accuracyRecordTransStep :: batchWriteSteps ++ streamingWriteSteps
+          accuracyRecordTransStep :: Nil
       }
     }
   }
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQSteps.scala
index 3df4a12..7312f29 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQSteps.scala
@@ -81,7 +81,7 @@
         s"SELECT ${selClause} FROM `${sourceName}`"
       }
       val sourceAliasTransStep =
-        SparkSqlTransformStep(sourceAliasTableName, sourceAliasSql, emptyMap, true)
+        SparkSqlTransformStep(sourceAliasTableName, sourceAliasSql, emptyMap, None, true)
 
       // 2. incomplete record
       val incompleteRecordsTableName = "__incompleteRecords"
@@ -91,15 +91,17 @@
       val incompleteRecordsSql =
         s"SELECT * FROM `${sourceAliasTableName}` WHERE ${incompleteWhereClause}"
 
-      val incompleteRecordTransStep =
-        SparkSqlTransformStep(incompleteRecordsTableName, incompleteRecordsSql, emptyMap, true)
-      incompleteRecordTransStep.parentSteps += sourceAliasTransStep
       val incompleteRecordWriteStep = {
         val rwName =
           ruleParam.getOutputOpt(RecordOutputType).flatMap(_.getNameOpt)
             .getOrElse(incompleteRecordsTableName)
         RecordWriteStep(rwName, incompleteRecordsTableName)
       }
+      val incompleteRecordTransStep =
+        SparkSqlTransformStep(incompleteRecordsTableName, incompleteRecordsSql, emptyMap,
+          Some(incompleteRecordWriteStep), true)
+      incompleteRecordTransStep.parentSteps += sourceAliasTransStep
+
 
       // 3. incomplete count
       val incompleteCountTableName = "__incompleteCount"
@@ -149,21 +151,19 @@
              |ON `${totalCountTableName}`.`${ConstantColumns.tmst}` = `${incompleteCountTableName}`.`${ConstantColumns.tmst}`
          """.stripMargin
       }
-      val completeTransStep = SparkSqlTransformStep(completeTableName, completeMetricSql, emptyMap)
-      completeTransStep.parentSteps += incompleteCountTransStep
-      completeTransStep.parentSteps += totalCountTransStep
       val completeWriteStep = {
         val metricOpt = ruleParam.getOutputOpt(MetricOutputType)
         val mwName = metricOpt.flatMap(_.getNameOpt).getOrElse(completeTableName)
         val flattenType = metricOpt.map(_.getFlatten).getOrElse(FlattenType.default)
         MetricWriteStep(mwName, completeTableName, flattenType)
       }
+      val completeTransStep =
+        SparkSqlTransformStep(completeTableName, completeMetricSql, emptyMap, Some(completeWriteStep))
+      completeTransStep.parentSteps += incompleteCountTransStep
+      completeTransStep.parentSteps += totalCountTransStep
 
       val transSteps = completeTransStep :: Nil
-      val writeSteps = incompleteRecordWriteStep :: completeWriteStep :: Nil
-
-      // full steps
-      transSteps ++ writeSteps
+      transSteps
     }
   }
 
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/DistinctnessExpr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/DistinctnessExpr2DQSteps.scala
index 0e2b10e..65460c3 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/DistinctnessExpr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/DistinctnessExpr2DQSteps.scala
@@ -102,7 +102,7 @@
         s"SELECT ${selClause} FROM `${sourceName}`"
       }
       val sourceAliasTransStep =
-        SparkSqlTransformStep(sourceAliasTableName, sourceAliasSql, emptyMap, true)
+        SparkSqlTransformStep(sourceAliasTableName, sourceAliasSql, emptyMap, None, true)
 
       // 2. total metric
       val totalTableName = "__totalMetric"
@@ -110,11 +110,12 @@
       val totalSql = {
         s"SELECT COUNT(*) AS `${totalColName}` FROM `${sourceAliasTableName}`"
       }
-      val totalTransStep = SparkSqlTransformStep(totalTableName, totalSql, emptyMap)
-      totalTransStep.parentSteps += sourceAliasTransStep
       val totalMetricWriteStep = {
         MetricWriteStep(totalColName, totalTableName, EntriesFlattenType, writeTimestampOpt)
       }
+      val totalTransStep =
+        SparkSqlTransformStep(totalTableName, totalSql, emptyMap, Some(totalMetricWriteStep))
+      totalTransStep.parentSteps += sourceAliasTransStep
 
       // 3. group by self
       val selfGroupTableName = "__selfGroup"
@@ -128,13 +129,12 @@
           """.stripMargin
       }
       val selfGroupTransStep =
-        SparkSqlTransformStep(selfGroupTableName, selfGroupSql, emptyMap, true)
+        SparkSqlTransformStep(selfGroupTableName, selfGroupSql, emptyMap, None, true)
       selfGroupTransStep.parentSteps += sourceAliasTransStep
 
       val transSteps1 = totalTransStep :: selfGroupTransStep :: Nil
-      val writeSteps1 = totalMetricWriteStep :: Nil
 
-      val ((transSteps2, writeSteps2), dupCountTableName) = procType match {
+      val (transSteps2, dupCountTableName) = procType match {
         case StreamingProcessType if (withOlderTable) =>
           // 4.0 update old data
           val targetDsUpdateWriteStep = DataSourceUpdateWriteStep(targetName, targetName)
@@ -208,14 +208,12 @@
              """.stripMargin
           }
           val finalDupCountTransStep =
-            SparkSqlTransformStep(finalDupCountTableName, finalDupCountSql, emptyMap, true)
+            SparkSqlTransformStep(finalDupCountTableName, finalDupCountSql, emptyMap, None, true)
           finalDupCountTransStep.parentSteps += groupTransStep
 
-          ((finalDupCountTransStep :: Nil, targetDsUpdateWriteStep :: Nil),
-            finalDupCountTableName)
+          (finalDupCountTransStep :: targetDsUpdateWriteStep :: Nil, finalDupCountTableName)
         case _ =>
-          ((selfGroupTransStep :: Nil, totalMetricWriteStep :: Nil),
-            selfGroupTableName)
+          (selfGroupTransStep :: Nil, selfGroupTableName)
       }
 
       // 8. distinct metric
@@ -227,16 +225,16 @@
            |FROM `${dupCountTableName}` WHERE `${ConstantColumns.distinct}`
          """.stripMargin
       }
-      val distTransStep = SparkSqlTransformStep(distTableName, distSql, emptyMap)
       val distMetricWriteStep = {
         MetricWriteStep(distColName, distTableName, EntriesFlattenType, writeTimestampOpt)
       }
+      val distTransStep =
+        SparkSqlTransformStep(distTableName, distSql, emptyMap, Some(distMetricWriteStep))
 
       val transSteps3 = distTransStep :: Nil
-      val writeSteps3 = distMetricWriteStep :: Nil
 
       val duplicationArrayName = details.getString(_duplicationArray, "")
-      val (transSteps4, writeSteps4) = if (duplicationArrayName.nonEmpty) {
+      val transSteps4 = if (duplicationArrayName.nonEmpty) {
         val recordEnable = details.getBoolean(_recordEnable, false)
         if (groupAliases.size > 0) {
           // with some group by requirement
@@ -278,12 +276,23 @@
                |WHERE NOT `${ConstantColumns.distinct}` OR `${ConstantColumns.rowNumber}` > 1
                """.stripMargin
           }
-          val dupItemsTransStep = SparkSqlTransformStep(dupItemsTableName, dupItemsSql, emptyMap)
-          dupItemsTransStep.parentSteps += rnTransStep
           val dupItemsWriteStep = {
             val rwName = ruleParam.getOutputOpt(RecordOutputType).flatMap(_.getNameOpt).getOrElse(dupItemsTableName)
             RecordWriteStep(rwName, dupItemsTableName, None, writeTimestampOpt)
           }
+          val dupItemsTransStep = {
+            if (recordEnable) {
+              SparkSqlTransformStep(
+                dupItemsTableName,
+                dupItemsSql,
+                emptyMap,
+                Some(dupItemsWriteStep)
+              )
+            } else {
+              SparkSqlTransformStep(dupItemsTableName, dupItemsSql, emptyMap)
+            }
+          }
+          dupItemsTransStep.parentSteps += rnTransStep
 
           // 12. group by dup Record metric
           val groupDupMetricTableName = "__groupDupMetric"
@@ -295,25 +304,22 @@
                |FROM `${dupItemsTableName}` GROUP BY ${groupSelClause}, `${dupColName}`
              """.stripMargin
           }
-          val groupDupMetricTransStep =
-            SparkSqlTransformStep(groupDupMetricTableName, groupDupMetricSql, emptyMap)
-          groupDupMetricTransStep.parentSteps += dupItemsTransStep
           val groupDupMetricWriteStep = {
             MetricWriteStep(duplicationArrayName,
               groupDupMetricTableName,
               ArrayFlattenType,
               writeTimestampOpt)
           }
+          val groupDupMetricTransStep =
+            SparkSqlTransformStep(
+              groupDupMetricTableName,
+              groupDupMetricSql,
+              emptyMap,
+              Some(groupDupMetricWriteStep)
+            )
+          groupDupMetricTransStep.parentSteps += dupItemsTransStep
 
-          val msteps = groupDupMetricTransStep :: Nil
-          val wsteps = if (recordEnable) {
-            dupItemsWriteStep :: groupDupMetricWriteStep :: Nil
-          } else {
-            groupDupMetricWriteStep :: Nil
-          }
-
-          (msteps, wsteps)
-
+          groupDupMetricTransStep :: Nil
         } else {
           // no group by requirement
           // 9. duplicate record
@@ -330,16 +336,25 @@
                |FROM `${dupCountTableName}` WHERE `${dupColName}` > 0
               """.stripMargin
           }
-          val dupRecordTransStep =
-            SparkSqlTransformStep(dupRecordTableName, dupRecordSql, emptyMap, true)
-
           val dupRecordWriteStep = {
             val rwName =
               ruleParam.getOutputOpt(RecordOutputType).flatMap(_.getNameOpt)
                 .getOrElse(dupRecordTableName)
-
             RecordWriteStep(rwName, dupRecordTableName, None, writeTimestampOpt)
           }
+          val dupRecordTransStep = {
+            if (recordEnable) {
+              SparkSqlTransformStep(
+                dupRecordTableName,
+                dupRecordSql,
+                emptyMap,
+                Some(dupRecordWriteStep),
+                true
+              )
+            } else {
+              SparkSqlTransformStep(dupRecordTableName, dupRecordSql, emptyMap, None, true)
+            }
+          }
 
           // 10. duplicate metric
           val dupMetricTableName = "__dupMetric"
@@ -350,8 +365,6 @@
                |FROM `${dupRecordTableName}` GROUP BY `${dupColName}`
               """.stripMargin
           }
-          val dupMetricTransStep = SparkSqlTransformStep(dupMetricTableName, dupMetricSql, emptyMap)
-          dupMetricTransStep.parentSteps += dupRecordTransStep
           val dupMetricWriteStep = {
             MetricWriteStep(
               duplicationArrayName,
@@ -360,22 +373,21 @@
               writeTimestampOpt
             )
           }
+          val dupMetricTransStep =
+            SparkSqlTransformStep(
+              dupMetricTableName,
+              dupMetricSql,
+              emptyMap,
+              Some(dupMetricWriteStep)
+            )
+          dupMetricTransStep.parentSteps += dupRecordTransStep
 
-          val msteps = dupMetricTransStep :: Nil
-          val wsteps = if (recordEnable) {
-            dupRecordWriteStep :: dupMetricWriteStep :: Nil
-          } else {
-            dupMetricWriteStep :: Nil
-          }
-
-          (msteps, wsteps)
+          dupMetricTransStep :: Nil
         }
-      } else (Nil, Nil)
+      } else Nil
 
       // full steps
-      transSteps1 ++ transSteps2 ++ transSteps3 ++ transSteps4 ++
-        writeSteps1 ++ writeSteps2 ++ writeSteps3 ++ writeSteps4
-
+      transSteps1 ++ transSteps2 ++ transSteps3 ++ transSteps4
     }
   }
 
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/Expr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/Expr2DQSteps.scala
index 492f4fd..e9a65b4 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/Expr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/Expr2DQSteps.scala
@@ -21,9 +21,10 @@
 import org.apache.griffin.measure.Loggable
 import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
 import org.apache.griffin.measure.configuration.enums._
-import org.apache.griffin.measure.context.{ContextId, DQContext, TimeRange}
+import org.apache.griffin.measure.context.DQContext
 import org.apache.griffin.measure.step.DQStep
 import org.apache.griffin.measure.step.builder.dsl.expr.Expr
+import org.apache.griffin.measure.step.write.{MetricWriteStep, RecordWriteStep, WriteStep}
 
 trait Expr2DQSteps extends Loggable with Serializable {
 
@@ -31,7 +32,6 @@
   protected val emptyMap = Map[String, Any]()
 
   def getDQSteps(): Seq[DQStep]
-
 }
 
 /**
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/ProfilingExpr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/ProfilingExpr2DQSteps.scala
index af493af..68ca2f4 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/ProfilingExpr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/ProfilingExpr2DQSteps.scala
@@ -97,14 +97,15 @@
           s"${fromClause} ${preGroupbyClause} ${groupbyClause} ${postGroupbyClause}"
       }
       val profilingName = ruleParam.getOutDfName()
-      val profilingTransStep = SparkSqlTransformStep(profilingName, profilingSql, details)
       val profilingMetricWriteStep = {
         val metricOpt = ruleParam.getOutputOpt(MetricOutputType)
         val mwName = metricOpt.flatMap(_.getNameOpt).getOrElse(ruleParam.getOutDfName())
         val flattenType = metricOpt.map(_.getFlatten).getOrElse(FlattenType.default)
         MetricWriteStep(mwName, profilingName, flattenType)
       }
-      profilingTransStep :: profilingMetricWriteStep :: Nil
+      val profilingTransStep =
+        SparkSqlTransformStep(profilingName, profilingSql, details, Some(profilingMetricWriteStep))
+      profilingTransStep :: Nil
     }
   }
 
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/TimelinessExpr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/TimelinessExpr2DQSteps.scala
index 03c2c8d..5a3acfb 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/TimelinessExpr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/TimelinessExpr2DQSteps.scala
@@ -104,7 +104,7 @@
         s"SELECT *, (`${etsColName}` - `${ConstantColumns.beginTs}`) AS `${latencyColName}` " +
           s"FROM `${inTimeTableName}`"
       }
-      val latencyTransStep = SparkSqlTransformStep(latencyTableName, latencySql, emptyMap, true)
+      val latencyTransStep = SparkSqlTransformStep(latencyTableName, latencySql, emptyMap, None, true)
       latencyTransStep.parentSteps += inTimeTransStep
 
       // 3. timeliness metric
@@ -129,27 +129,26 @@
              |GROUP BY `${ConstantColumns.tmst}`
            """.stripMargin
       }
-      val metricTransStep = SparkSqlTransformStep(metricTableName, metricSql, emptyMap)
-      metricTransStep.parentSteps += latencyTransStep
       val metricWriteStep = {
         val metricOpt = ruleParam.getOutputOpt(MetricOutputType)
         val mwName = metricOpt.flatMap(_.getNameOpt).getOrElse(ruleParam.getOutDfName())
         val flattenType = metricOpt.map(_.getFlatten).getOrElse(FlattenType.default)
         MetricWriteStep(mwName, metricTableName, flattenType)
       }
+      val metricTransStep =
+        SparkSqlTransformStep(metricTableName, metricSql, emptyMap, Some(metricWriteStep))
+      metricTransStep.parentSteps += latencyTransStep
 
       // current steps
       val transSteps1 = metricTransStep :: Nil
-      val writeSteps1 = metricWriteStep :: Nil
 
       // 4. timeliness record
-      val (transSteps2, writeSteps2) = TimeUtil.milliseconds(details.getString(_threshold, "")) match {
+      val transSteps2 = TimeUtil.milliseconds(details.getString(_threshold, "")) match {
         case Some(tsh) =>
           val recordTableName = "__lateRecords"
           val recordSql = {
             s"SELECT * FROM `${latencyTableName}` WHERE `${latencyColName}` > ${tsh}"
           }
-          val recordTransStep = SparkSqlTransformStep(recordTableName, recordSql, emptyMap)
           val recordWriteStep = {
             val rwName =
               ruleParam.getOutputOpt(RecordOutputType).flatMap(_.getNameOpt)
@@ -157,12 +156,16 @@
 
             RecordWriteStep(rwName, recordTableName, None)
           }
-          (recordTransStep :: Nil, recordWriteStep :: Nil)
-        case _ => (Nil, Nil)
+          val recordTransStep =
+            SparkSqlTransformStep(recordTableName, recordSql, emptyMap, Some(recordWriteStep))
+          recordTransStep.parentSteps += latencyTransStep
+
+          recordTransStep :: Nil
+        case _ => Nil
       }
 
       // 5. ranges
-      val (transSteps3, writeSteps3) = TimeUtil.milliseconds(details.getString(_stepSize, "")) match {
+      val transSteps3 = TimeUtil.milliseconds(details.getString(_stepSize, "")) match {
         case Some(stepSize) =>
           // 5.1 range
           val rangeTableName = "__range"
@@ -174,6 +177,7 @@
              """.stripMargin
           }
           val rangeTransStep = SparkSqlTransformStep(rangeTableName, rangeSql, emptyMap)
+          rangeTransStep.parentSteps += latencyTransStep
 
           // 5.2 range metric
           val rangeMetricTableName = "__rangeMetric"
@@ -190,20 +194,20 @@
                  |FROM `${rangeTableName}` GROUP BY `${ConstantColumns.tmst}`, `${stepColName}`
                 """.stripMargin
           }
-          val rangeMetricTransStep =
-            SparkSqlTransformStep(rangeMetricTableName, rangeMetricSql, emptyMap)
-          rangeMetricTransStep.parentSteps += rangeTransStep
           val rangeMetricWriteStep = {
             MetricWriteStep(stepColName, rangeMetricTableName, ArrayFlattenType)
           }
+          val rangeMetricTransStep =
+            SparkSqlTransformStep(rangeMetricTableName, rangeMetricSql, emptyMap, Some(rangeMetricWriteStep))
+          rangeMetricTransStep.parentSteps += rangeTransStep
 
-          (rangeMetricTransStep :: Nil, rangeMetricWriteStep :: Nil)
-        case _ => (Nil, Nil)
+          rangeMetricTransStep :: Nil
+        case _ => Nil
       }
 
       // 6. percentiles
       val percentiles = getPercentiles(details)
-      val (transSteps4, writeSteps4) = if (percentiles.size > 0) {
+      val transSteps4 = if (percentiles.size > 0) {
         val percentileTableName = "__percentile"
         val percentileColName = details.getStringOrKey(_percentileColPrefix)
         val percentileCols = percentiles.map { pct =>
@@ -217,19 +221,18 @@
              |FROM `${latencyTableName}`
             """.stripMargin
         }
-        val percentileTransStep =
-          SparkSqlTransformStep(percentileTableName, percentileSql, emptyMap)
-
         val percentileWriteStep = {
           MetricWriteStep(percentileTableName, percentileTableName, DefaultFlattenType)
         }
+        val percentileTransStep =
+          SparkSqlTransformStep(percentileTableName, percentileSql, emptyMap, Some(percentileWriteStep))
+        percentileTransStep.parentSteps += latencyTransStep
 
-        (percentileTransStep :: Nil, percentileWriteStep :: Nil)
-      } else (Nil, Nil)
+        percentileTransStep :: Nil
+      } else Nil
 
       // full steps
-      transSteps1 ++ transSteps2 ++ transSteps3 ++ transSteps4 ++
-        writeSteps1 ++ writeSteps2 ++ writeSteps3 ++ writeSteps4
+      transSteps1 ++ transSteps2 ++ transSteps3 ++ transSteps4
     }
   }
 
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/UniquenessExpr2DQSteps.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/UniquenessExpr2DQSteps.scala
index 7f259ea..a19b35c 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/UniquenessExpr2DQSteps.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/UniquenessExpr2DQSteps.scala
@@ -117,7 +117,7 @@
         s"SELECT ${groupSelClause}, (COUNT(*) - 1) AS `${dupColName}` " +
           s"FROM `${joinedTableName}` GROUP BY ${groupSelClause}"
       }
-      val groupTransStep = SparkSqlTransformStep(groupTableName, groupSql, emptyMap, true)
+      val groupTransStep = SparkSqlTransformStep(groupTableName, groupSql, emptyMap, None, true)
       groupTransStep.parentSteps += joinedTransStep
 
       // 5. total metric
@@ -131,8 +131,9 @@
              |FROM `${sourceName}` GROUP BY `${ConstantColumns.tmst}`
            """.stripMargin
       }
-      val totalTransStep = SparkSqlTransformStep(totalTableName, totalSql, emptyMap)
       val totalMetricWriteStep = MetricWriteStep(totalColName, totalTableName, EntriesFlattenType)
+      val totalTransStep =
+        SparkSqlTransformStep(totalTableName, totalSql, emptyMap, Some(totalMetricWriteStep))
 
       // 6. unique record
       val uniqueRecordTableName = "__uniqueRecord"
@@ -155,24 +156,21 @@
              |FROM `${uniqueRecordTableName}` GROUP BY `${ConstantColumns.tmst}`
            """.stripMargin
       }
-      val uniqueTransStep = SparkSqlTransformStep(uniqueTableName, uniqueSql, emptyMap)
-      uniqueTransStep.parentSteps += uniqueRecordTransStep
-
       val uniqueMetricWriteStep =
         MetricWriteStep(uniqueColName, uniqueTableName, EntriesFlattenType)
+      val uniqueTransStep =
+        SparkSqlTransformStep(uniqueTableName, uniqueSql, emptyMap, Some(uniqueMetricWriteStep))
+      uniqueTransStep.parentSteps += uniqueRecordTransStep
 
       val transSteps1 = totalTransStep :: uniqueTransStep :: Nil
-      val writeSteps1 = totalMetricWriteStep :: uniqueMetricWriteStep :: Nil
 
       val duplicationArrayName = details.getString(_duplicationArray, "")
-      val (transSteps2, writeSteps2) = if (duplicationArrayName.nonEmpty) {
+      val transSteps2 = if (duplicationArrayName.nonEmpty) {
         // 8. duplicate record
         val dupRecordTableName = "__dupRecords"
         val dupRecordSql = {
           s"SELECT * FROM `${groupTableName}` WHERE `${dupColName}` > 0"
         }
-        val dupRecordTransStep =
-          SparkSqlTransformStep(dupRecordTableName, dupRecordSql, emptyMap, true)
 
         val dupRecordWriteStep = {
           val rwName =
@@ -181,6 +179,8 @@
 
           RecordWriteStep(rwName, dupRecordTableName)
         }
+        val dupRecordTransStep =
+          SparkSqlTransformStep(dupRecordTableName, dupRecordSql, emptyMap, Some(dupRecordWriteStep), true)
 
         // 9. duplicate metric
         val dupMetricTableName = "__dupMetric"
@@ -201,18 +201,22 @@
              |GROUP BY ${dupMetricGroupbyClause}
           """.stripMargin
         }
-        val dupMetricTransStep = SparkSqlTransformStep(dupMetricTableName, dupMetricSql, emptyMap)
-        dupMetricTransStep.parentSteps += dupRecordTransStep
         val dupMetricWriteStep = {
           MetricWriteStep(duplicationArrayName, dupMetricTableName, ArrayFlattenType)
         }
+        val dupMetricTransStep =
+          SparkSqlTransformStep(dupMetricTableName,
+            dupMetricSql,
+            emptyMap,
+            Some(dupMetricWriteStep)
+          )
+        dupMetricTransStep.parentSteps += dupRecordTransStep
 
-        (dupMetricTransStep :: Nil,
-          dupRecordWriteStep :: dupMetricWriteStep :: Nil)
-      } else (Nil, Nil)
+        dupMetricTransStep :: Nil
+      } else Nil
 
       // full steps
-      transSteps1 ++ transSteps2 ++ writeSteps1 ++ writeSteps2
+      transSteps1 ++ transSteps2
     }
   }
 
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
index b07595a..c393706 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/transform/DataFrameOpsTransformStep.scala
@@ -19,14 +19,16 @@
 package org.apache.griffin.measure.step.transform
 
 import org.apache.griffin.measure.context.DQContext
+import org.apache.griffin.measure.step.write.WriteStep
 
 /**
   * data frame ops transform step
   */
-case class DataFrameOpsTransformStep(name: String,
+case class DataFrameOpsTransformStep[T <: WriteStep](name: String,
                                      inputDfName: String,
                                      rule: String,
                                      details: Map[String, Any],
+                                     writeStepOpt: Option[T] = None,
                                      cache: Boolean = false
                                     ) extends TransformStep {
 
@@ -43,7 +45,10 @@
       }
       if (cache) context.dataFrameCache.cacheDataFrame(name, df)
       context.runTimeTableRegister.registerTable(name, df)
-      true
+      writeStepOpt match {
+        case Some(writeStep) => writeStep.execute(context)
+        case None => true
+      }
     } catch {
       case e: Throwable =>
         error(s"run data frame ops [ ${rule} ] error: ${e.getMessage}", e)
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
index 59ea822..00edf07 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/transform/SparkSqlTransformStep.scala
@@ -19,23 +19,27 @@
 package org.apache.griffin.measure.step.transform
 
 import org.apache.griffin.measure.context.DQContext
+import org.apache.griffin.measure.step.write.WriteStep
 
 /**
   * spark sql transform step
   */
-case class SparkSqlTransformStep(name: String,
-                                 rule: String,
-                                 details: Map[String, Any],
-                                 cache: Boolean = false
-                                ) extends TransformStep {
-
+case class SparkSqlTransformStep[T <: WriteStep](name: String,
+                                                 rule: String,
+                                                 details: Map[String, Any],
+                                                 writeStepOpt: Option[T] = None,
+                                                 cache: Boolean = false
+                                                ) extends TransformStep {
   def doExecute(context: DQContext): Boolean = {
     val sqlContext = context.sqlContext
     try {
       val df = sqlContext.sql(rule)
       if (cache) context.dataFrameCache.cacheDataFrame(name, df)
       context.runTimeTableRegister.registerTable(name, df)
-      true
+      writeStepOpt match {
+        case Some(writeStep) => writeStep.execute(context)
+        case None => true
+      }
     } catch {
       case e: Throwable =>
         error(s"run spark sql [ ${rule} ] error: ${e.getMessage}", e)
diff --git a/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala b/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala
index e640c45..5314669 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala
@@ -19,15 +19,14 @@
 package org.apache.griffin.measure.step
 
 import com.holdenkarau.spark.testing.DataFrameSuiteBase
+import org.scalatest._
 
+import org.apache.griffin.measure.Loggable
 import org.apache.griffin.measure.configuration.enums.BatchProcessType
 import org.apache.griffin.measure.context.ContextId
 import org.apache.griffin.measure.context.DQContext
-import org.apache.griffin.measure.Loggable
 import org.apache.griffin.measure.step.transform.TransformStep
 
-import org.scalatest._
-
 class TransformStepTest extends FlatSpec with Matchers with DataFrameSuiteBase with Loggable {
 
   case class DualTransformStep(name: String,