You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@seatunnel.apache.org by ki...@apache.org on 2022/08/31 15:12:48 UTC
[incubator-seatunnel] branch dev updated: Add empty dataset verification to avoid the exception caused by running transformerprocess on an empty dataset (#2578)
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 d82470c68 Add empty dataset verification to avoid the exception caused by running transformerprocess on an empty dataset (#2578)
d82470c68 is described below
commit d82470c6858f1e5770bf3eeda71a216edc29814e
Author: miaoze8 <st...@qq.com>
AuthorDate: Wed Aug 31 23:12:43 2022 +0800
Add empty dataset verification to avoid the exception caused by running transformerprocess on an empty dataset (#2578)
Co-authored-by: ze.miao <ze...@pcitc.com>
---
.../apache/seatunnel/spark/stream/SparkStreamingExecution.scala | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
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 3ad85d3c8..afb62667f 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
@@ -48,9 +48,12 @@ class SparkStreamingExecution(sparkEnvironment: SparkEnvironment)
dataset)
}
var ds = dataset
- for (tf <- transforms) {
- ds = SparkEnvironment.transformProcess(sparkEnvironment, tf, ds)
- SparkEnvironment.registerTransformTempView(tf, ds)
+
+ if (ds.take(1).length > 0) {
+ for (tf <- transforms) {
+ ds = SparkEnvironment.transformProcess(sparkEnvironment, tf, ds)
+ SparkEnvironment.registerTransformTempView(tf, ds)
+ }
}
source.beforeOutput()