[incubator-seatunnel] branch dev updated: [Bug][Spark] When have empty data source, transform not execute (#1509)

[kirs]

This is an automated email from the ASF dual-hosted git repository.

kirs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new c65067d  [Bug][Spark] When have empty data source, transform not 
execute (#1509)
c65067d is described below

commit c65067d95a2da54116d3278b8e7ddfb912c77354
Author: TrickyZerg <32387433+benj...@users.noreply.github.com>
AuthorDate: Sat Mar 19 20:54:00 2022 +0800

    [Bug][Spark] When have empty data source, transform not execute (#1509)
---
 .../java/org/apache/seatunnel/spark/batch/SparkBatchExecution.java  | 6 ++----
 .../org/apache/seatunnel/spark/stream/SparkStreamingExecution.scala | 6 ++----
 2 files changed, 4 insertions(+), 8 deletions(-)

diff --git 
a/seatunnel-apis/seatunnel-api-spark/src/main/java/org/apache/seatunnel/spark/batch/SparkBatchExecution.java
 
b/seatunnel-apis/seatunnel-api-spark/src/main/java/org/apache/seatunnel/spark/batch/SparkBatchExecution.java
index 5674c28..0eb6568 100644
--- 
a/seatunnel-apis/seatunnel-api-spark/src/main/java/org/apache/seatunnel/spark/batch/SparkBatchExecution.java
+++ 
b/seatunnel-apis/seatunnel-api-spark/src/main/java/org/apache/seatunnel/spark/batch/SparkBatchExecution.java
@@ -96,10 +96,8 @@ public class SparkBatchExecution implements 
Execution<SparkBatchSource, BaseSpar
         if (!sources.isEmpty()) {
             Dataset<Row> ds = sources.get(0).getData(environment);
             for (BaseSparkTransform transform : transforms) {
-                if (ds.takeAsList(1).size() > 0) {
-                    ds = transformProcess(environment, transform, ds);
-                    registerTransformTempView(transform, ds);
-                }
+                ds = transformProcess(environment, transform, ds);
+                registerTransformTempView(transform, ds);
             }
             for (SparkBatchSink sink : sinks) {
                 sinkProcess(environment, sink, ds);
diff --git 
a/seatunnel-apis/seatunnel-api-spark/src/main/scala/org/apache/seatunnel/spark/stream/SparkStreamingExecution.scala
 
b/seatunnel-apis/seatunnel-api-spark/src/main/scala/org/apache/seatunnel/spark/stream/SparkStreamingExecution.scala
index 2ffaf15..df75334 100644
--- 
a/seatunnel-apis/seatunnel-api-spark/src/main/scala/org/apache/seatunnel/spark/stream/SparkStreamingExecution.scala
+++ 
b/seatunnel-apis/seatunnel-api-spark/src/main/scala/org/apache/seatunnel/spark/stream/SparkStreamingExecution.scala
@@ -51,10 +51,8 @@ class SparkStreamingExecution(sparkEnvironment: 
SparkEnvironment)
         }
         var ds = dataset
         for (tf <- transforms) {
-          if (ds.take(1).length > 0) {
-            ds = SparkBatchExecution.transformProcess(sparkEnvironment, tf, ds)
-            SparkBatchExecution.registerTransformTempView(tf, ds)
-          }
+          ds = SparkBatchExecution.transformProcess(sparkEnvironment, tf, ds)
+          SparkBatchExecution.registerTransformTempView(tf, ds)
         }
 
         source.beforeOutput()