Merge pull request #362 from InterestingLab/rickyhuo.fixbug.streaming

Fix a bug of streaming pipeline when register table view
Wzhipeng · Aug 12, 2019 · 433560a · 433560a
2 parents 167d639 + 3f79832
commit 433560a
Show file tree

Hide file tree

Showing 4 changed files with 65 additions and 6 deletions.
diff --git a/config/structuredstreaming.conf.template b/config/structuredstreaming.conf.template
@@ -14,6 +14,7 @@ input {
   kafkaStream {
     consumer.bootstrap.servers = "127.0.0.1:9092"
     topics = "waterdrop_input"
+    consumer.group.id = "waterdrop_input_group_id"
     result_table_name = "waterdrop_message"
   }
 }

diff --git a/waterdrop-core/src/main/scala/io/github/interestinglab/waterdrop/Waterdrop.scala b/waterdrop-core/src/main/scala/io/github/interestinglab/waterdrop/Waterdrop.scala
@@ -157,7 +157,7 @@ object Waterdrop extends Logging {
         for (f <- filters) {
           if (ds.take(1).length > 0) {
             ds = filterProcess(sparkSession, f, ds)
-            registerFilterTempView(f, ds)
+            registerStreamingFilterTempView(f, ds)
           }
         }
 
@@ -293,7 +293,7 @@ object Waterdrop extends Logging {
     }
   }
 
-  private[waterdrop] def registerInputTempView(input: Plugin, ds: Dataset[Row]): Unit = {
+  private[waterdrop] def registerInputTempView(input: BaseStaticInput, ds: Dataset[Row]): Unit = {
     val config = input.getConfig()
     config.hasPath("table_name") || config.hasPath("result_table_name") match {
       case true => {
@@ -316,6 +316,29 @@ object Waterdrop extends Logging {
     }
   }
 
+  private[waterdrop] def registerInputTempView(input: BaseStreamingInput[Any], ds: Dataset[Row]): Unit = {
+    val config = input.getConfig()
+    config.hasPath("table_name") || config.hasPath("result_table_name") match {
+      case true => {
+        val tableName = config.hasPath("table_name") match {
+          case true => {
+            @deprecated
+            val oldTableName = config.getString("table_name")
+            oldTableName
+          }
+          case false => config.getString("result_table_name")
+        }
+        registerStreamingTempView(tableName, ds)
+      }
+
+      case false => {
+        throw new ConfigRuntimeException(
+          "Plugin[" + input.name + "] must be registered as dataset/table, please set \"result_table_name\" config")
+
+      }
+    }
+  }
+
   private[waterdrop] def registerFilterTempView(plugin: Plugin, ds: Dataset[Row]): Unit = {
     val config = plugin.getConfig()
     if (config.hasPath("result_table_name")) {
@@ -324,6 +347,14 @@ object Waterdrop extends Logging {
     }
   }
 
+  private[waterdrop] def registerStreamingFilterTempView(plugin: BaseFilter, ds: Dataset[Row]): Unit = {
+    val config = plugin.getConfig()
+    if (config.hasPath("result_table_name")) {
+      val tableName = config.getString("result_table_name")
+      registerStreamingTempView(tableName, ds)
+    }
+  }
+
   private[waterdrop] def registerTempView(tableName: String, ds: Dataset[Row]): Unit = {
     viewTableMap.contains(tableName) match {
       case true =>
@@ -337,6 +368,10 @@ object Waterdrop extends Logging {
     }
   }
 
+  private[waterdrop] def registerStreamingTempView(tableName: String, ds: Dataset[Row]): Unit = {
+    ds.createOrReplaceTempView(tableName)
+  }
+
   private[waterdrop] def deployModeCheck(): Unit = {
     Common.getDeployMode match {
       case Some(m) => {

diff --git a/...core/src/main/scala/io/github/interestinglab/waterdrop/WaterdropStructuredStreaming.scala b/...core/src/main/scala/io/github/interestinglab/waterdrop/WaterdropStructuredStreaming.scala
@@ -95,7 +95,7 @@ object WaterdropStructuredStreaming extends Logging {
 
     val datasetList = structuredStreamingInputs.map(p => {
       val ds = p.getDataset(sparkSession)
-      Waterdrop.registerInputTempView(p, ds)
+      registerInputTempView(p, ds)
       ds
     })
 
@@ -107,7 +107,7 @@ object WaterdropStructuredStreaming extends Logging {
     var ds: Dataset[Row] = datasetList.get(0)
     for (f <- filters) {
       ds = Waterdrop.filterProcess(sparkSession, f, ds)
-      Waterdrop.registerFilterTempView(f, ds)
+      Waterdrop.registerStreamingFilterTempView(f, ds)
     }
 
     var streamingQueryList = List[StreamingQuery]()
@@ -122,6 +122,29 @@ object WaterdropStructuredStreaming extends Logging {
     }
   }
 
+  private def registerInputTempView(input: BaseStructuredStreamingInput, ds: Dataset[Row]): Unit = {
+    val config = input.getConfig()
+    config.hasPath("table_name") || config.hasPath("result_table_name") match {
+      case true => {
+        val tableName = config.hasPath("table_name") match {
+          case true => {
+            @deprecated
+            val oldTableName = config.getString("table_name")
+            oldTableName
+          }
+          case false => config.getString("result_table_name")
+        }
+        Waterdrop.registerStreamingTempView(tableName, ds)
+      }
+
+      case false => {
+        throw new ConfigRuntimeException(
+          "Plugin[" + input.name + "] must be registered as dataset/table, please set \"result_table_name\" config")
+
+      }
+    }
+  }
+
   private def structuredStreamingOutputProcess(
     sparkSession: SparkSession,
     output: BaseStructuredStreamingOutputIntra,

diff --git a/waterdrop-core/src/main/scala/io/github/interestinglab/waterdrop/filter/Sql.scala b/waterdrop-core/src/main/scala/io/github/interestinglab/waterdrop/filter/Sql.scala
@@ -25,9 +25,9 @@ class Sql extends BaseFilter {
   }
 
   override def checkConfig(): (Boolean, String) = {
-    conf.hasPath("sql") match {
+    conf.hasPath("table_name") match {
       case true => {
-        if (conf.hasPath("sql")) {
+        if (conf.hasPath("table_name")) {
           logWarning("parameter [table_name] is deprecated since 1.4")
         }
         (true, "")