Merge pull request #20 from databrickslabs/dev

Added extra filter for change data feed.
databrickslabs · Oct 5, 2021 · fa2d73d · fa2d73d
2 parents 79af73c + 93955e6
commit fa2d73d
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/src/main/scala/com/databricks/labs/deltaoms/common/OMSOperations.scala b/src/main/scala/com/databricks/labs/deltaoms/common/OMSOperations.scala
@@ -352,6 +352,7 @@ trait OMSOperations extends Serializable with SparkSettings with Logging with Sc
       .option("readChangeFeed", "true")
       .option("startingVersion", lastProcessedVersion + 1)
       .load(s"$rawActionsTablePath")
+      .filter("""_change_type IN ("insert", "update_postimage")""")
   }
 
   def processCommitInfoFromRawActions(rawActions: DataFrame,

diff --git a/src/main/scala/com/databricks/labs/deltaoms/configuration/SparkSettings.scala b/src/main/scala/com/databricks/labs/deltaoms/configuration/SparkSettings.scala
@@ -37,7 +37,6 @@ trait SparkSettings extends Serializable with ConfigurationSettings {
       spark.conf.set("spark.databricks.labs.deltaoms.version", value = Schemas.OMS_VERSION)
       spark
   }
-
-  def spark: SparkSession = SparkSession.active
+  def spark: SparkSession = sparkSession
 
 }