fix SQLCompositor error when checkpoint enabled

allwefantasy · allwefantasy · commit 2d1d8a3d69b1 · 2016-11-16T11:33:53.000+08:00
diff --git a/src/main/java/streaming/core/compositor/spark/streaming/transformation/SQLCompositor.scala b/src/main/java/streaming/core/compositor/spark/streaming/transformation/SQLCompositor.scala
@@ -39,6 +39,7 @@ class SQLCompositor[T] extends Compositor[T] with CompositorHelper {
 
     require(sql.isDefined, "please set sql  by variable `sql` in config file")
     val _sql = translateSQL(sql.get,params)
+    val _outputTableName = outputTableName
     if (params.containsKey(TABLE)) {
 
       //parent compositor is  tableCompositor
@@ -47,7 +48,7 @@ class SQLCompositor[T] extends Compositor[T] with CompositorHelper {
       params.put(FUNC, (df: Any) => {
         val sqlContext = func(df)
         val newDF = sqlContext.sql(_sql)
-        outputTableName match {
+        _outputTableName match {
           case Some(tableName) =>
             newDF.registerTempTable(tableName)
           case None =>
@@ -68,7 +69,7 @@ class SQLCompositor[T] extends Compositor[T] with CompositorHelper {
         }
 
         val newDF = func(oldDF).sqlContext.sql(_sql)
-        outputTableName match {
+        _outputTableName match {
           case Some(tableName) =>
             newDF.registerTempTable(tableName)
           case None =>
diff --git a/src/main/java/streaming/core/compositor/spark/transformation/SQLCompositor.scala b/src/main/java/streaming/core/compositor/spark/transformation/SQLCompositor.scala
@@ -39,14 +39,15 @@ class SQLCompositor[T] extends Compositor[T] with CompositorHelper {
 
     require(sql.isDefined, "please set sql  by variable `sql` in config file")
     val _sql = translateSQL(sql.get,params)
+    val _outputTableName = outputTableName
     if (params.containsKey(TABLE)) {
       //parent compositor is  tableCompositor
 
       val func = params.get(TABLE).asInstanceOf[(Any) => SQLContext]
       params.put(FUNC, (rddOrDF: Any) => {
         val sqlContext = func(rddOrDF)
         val newDF = sqlContext.sql(_sql)
-        outputTableName match {
+        _outputTableName match {
           case Some(tableName) =>
             newDF.registerTempTable(tableName)
           case None =>
@@ -59,7 +60,7 @@ class SQLCompositor[T] extends Compositor[T] with CompositorHelper {
       val func = params.get(FUNC).asInstanceOf[(DataFrame) => DataFrame]
       params.put(FUNC, (df: DataFrame) => {
         val newDF = func(df).sqlContext.sql(_sql)
-        outputTableName match {
+        _outputTableName match {
           case Some(tableName) =>
             newDF.registerTempTable(tableName)
           case None =>
