Optimize batch pipeline: no need in cache w/o deadletters

Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
feast-dev · Apr 15, 2021 · 691cb45 · 691cb45
1 parent 1b24bac
commit 691cb45
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala
@@ -62,7 +62,11 @@ object BatchPipeline extends BasePipeline {
         )
     }
 
-    val projected = input.select(projection: _*).cache()
+    val projected = if (config.deadLetterPath.nonEmpty) {
+      input.select(projection: _*).cache()
+    } else {
+      input.select(projection: _*)
+    }
 
     TypeCheck.allTypesMatch(projected.schema, featureTable) match {
       case Some(error) =>