Validating Spark DataFrames: dropna, fillna, replace and Quarantine Filters

1.Dropping incomplete rows with dropna()

Validating a DataFrame usually starts with missing values. The DataFrame.na property returns a DataFrameNaFunctions object with three methods: drop, fill and replace. Each has a top-level equivalent on the DataFrame, for example "DataFrame.dropna and DataFrameNaFunctions.drop are aliases of each other." The full signature is dropna(how='any', thresh=None, subset=None).

from pyspark.sql import Row
df = spark.createDataFrame([
    Row(age=10, height=80.0, name="Alice"),
    Row(age=5, height=float("nan"), name="Bob"),
    Row(age=None, height=None, name="Tom"),
    Row(age=None, height=float("nan"), name=None),
])

df.na.drop().show()
# +---+------+-----+
# |age|height| name|
# +---+------+-----+
# | 10|  80.0|Alice|
# +---+------+-----+

df.na.drop(how='all').show()
# +----+------+-----+
# | age|height| name|
# +----+------+-----+
# |  10|  80.0|Alice|
# |   5|   NaN|  Bob|
# |NULL|  NULL|  Tom|
# +----+------+-----+

df.na.drop(thresh=2).show()
# +---+------+-----+
# |age|height| name|
# +---+------+-----+
# | 10|  80.0|Alice|
# |  5|   NaN|  Bob|
# +---+------+-----+

With how='all', Tom is kept because his name isn't null. The last row is dropped because its age and name are null and its height is NaN, which shows that NaN counts as missing here too. With thresh=2, a row needs at least two non-null values to stay. Tom has one, so he is dropped, and Bob, with age and name present, is kept. thresh doesn't combine with how. The reference says: "If specified, drop rows that have less than thresh non-null values. This overwrites the how parameter."

df.na.drop( ? =2).show()

Sources1

2.Repairing values with fillna() and na.replace()

Dropping rows throws data away. Often it is better to fill in a default value. Here too, "DataFrame.fillna and DataFrameNaFunctions.fill are aliases of each other." The value argument can be an int, float, string or bool, or a dict that maps column names to replacement values. With a single value, the type of that value decides which columns get filled.

df = spark.createDataFrame([
    (10, 80.5, "Alice", None),
    (5, None, "Bob", None),
    (None, None, "Tom", None),
    (None, None, None, True)],
    schema=["age", "height", "name", "bool"])

df.na.fill(50).show()
# +---+------+-----+----+
# |age|height| name|bool|
# +---+------+-----+----+
# | 10|  80.5|Alice|NULL|
# |  5|  50.0|  Bob|NULL|
# | 50|  50.0|  Tom|NULL|
# | 50|  50.0| NULL|true|
# +---+------+-----+----+

age and height were filled, and height shows the value as 50.0. The string column name and the boolean column bool still contain NULL, because "Columns specified in subset that do not have matching data types are ignored." The same rule works the other way: in the reference, df.na.fill(False) changes only bool. To set different values for different columns in one call, pass a dict such as {'age': 50, 'name': 'unknown'}. In that case "If the value is a dict, then subset is ignored", because the dict already names the columns.

fill only changes nulls. To change values that are present but wrong, such as inconsistent codes, use na.replace(to_replace, value, subset). It "Returns a new DataFrame replacing a value with another value." The last argument limits the replacement to particular columns.

df.na.replace(['Alice', 'Bob'], ['A', 'B'], 'name').show()

+----+------+----+
| age|height|name|
+----+------+----+
|  10|    80|   A|
|   5|  NULL|   B|
|NULL|    10| Tom|
+----+------+----+

Sources23

3.Rule-based validation: filters, quarantine and placeholders

Missing values are only one kind of problem. Business rules, such as "quantity can't be negative" or "an event can't be in the future", are checked with filter (also available as where). Its condition is "A Column of BooleanType or a string of SQL expressions." Column methods such as isNull(), isNotNull() and isNaN() produce those boolean columns, and you can combine conditions with &, as in df.filter((df.age > 3) & (df.subject == "Physics")).

from pyspark.sql import Row
df = spark.createDataFrame([Row(name='Tom', height=80), Row(name='Alice', height=None)])
df.filter(df.height.isNull()).collect()

# [Row(name='Alice', height=None)]

Silently discarding bad rows hides problems, so the Databricks validation guide uses quarantine instead: "You can use filters and WHERE clauses to define custom logic that quarantines bad records and prevents them from propagating to downstream tables." One query writes the rows that pass the rules to the clean table. A second query, with the opposite condition, writes the rows that fail to a quarantine table where someone can inspect them.

INSERT INTO silver_table
  SELECT * FROM bronze_table
  WHERE event_timestamp <= current_time AND quantity >= 0;

INSERT INTO quarantine_table
  SELECT * FROM bronze_table
  WHERE event_timestamp > current_time OR quantity < 0;

The guide makes a point about how to write the results: "Databricks recommends always processing filtered data as a separate write operation, especially when using Structured Streaming." It warns against writing both outputs from a single .foreachBatch. Some invalid values should be repaired rather than quarantined. Suppose an upstream system can't encode NULL, so "the placeholder value -1 is used to represent missing data." You can convert -1 to NULL once in the pipeline, so downstream queries don't each need to filter it out.

INSERT INTO silver_table
  SELECT
    * EXCEPT weight,
    CASE
      WHEN weight = -1 THEN NULL
      ELSE weight
    END AS weight
  FROM bronze_table;

The DataFrame API has the same construct in when(...).otherwise(...), which Spark displays as a CASE WHEN column. If you leave out otherwise, rows that match no condition get null: "If otherwise() is not invoked, None is returned for unmatched conditions." Rules that must hold for every row of a stored table can also be declared on the table itself with a SQL CHECK constraint, which "allows you to define a condition that must be true for every row in the table."

from pyspark.sql import functions as dbf
df = spark.range(3)
df.select("*", dbf.when(df['id'] == 2, 3).otherwise(4)).show()

Sources456

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Passing both how='any' and thresh=2 to dropna applies both rules.Why is that wrong?

   When thresh is set, it overrides how. Rows are dropped only when they have fewer than thresh non-null values.

   Covered in Dropping incomplete rows with dropna()

2. 2.df.na.fill(50) replaces every null in the DataFrame, including nulls in string and boolean columns.Why is that wrong?

   Only columns whose type matches the fill value are filled. In the reference output, name and bool still contain NULL after fill(50).

   Covered in Repairing values with fillna() and na.replace()

3. 3.fillna({'age': 50}, subset=['name']) limits the fill to the name column.Why is that wrong?

   When value is a dict, subset is ignored. The dict's keys alone decide which columns are filled.

   Covered in Repairing values with fillna() and na.replace()

4. 4.The recommended way to split valid and quarantined rows in a stream is a single .foreachBatch that writes to both tables.Why is that wrong?

   Databricks recommends writing the filtered data as separate write operations, because writing to several tables from .foreachBatch can produce inconsistent results.

   Covered in Rule-based validation: filters, quarantine and placeholders

Practise it for real

Run the reference dropna and fillna examples yourself and confirm how each option treats nulls and NaN

1. 1.Create the four-row age/height/name DataFrame from the dropna reference, including Bob's float('nan') height, and run df.na.drop().show().

   Why: This shows that the default how='any' treats NaN the same as null.

   You should see: Only the Alice row is left.

2. 2.On the same DataFrame, run df.na.drop(how='all').show() and then df.na.drop(thresh=2).show().

   Why: Comparing the two outputs shows the difference between 'every value missing' and 'too few values present'.

   You should see: how='all' keeps Alice, Bob and Tom. thresh=2 keeps Alice and Bob.

3. 3.Create the age/height/name/bool DataFrame from the fillna reference and run df.na.fill(50).show().

   Why: This checks which columns a single numeric fill value affects.

   You should see: age and height are filled with 50 (shown as 50.0 in height). name and bool still contain NULL.

4. 4.Run df.na.fill({'age': 50, 'name': 'unknown'}).show() on the same DataFrame.

   Why: A dict sets a separate fill value for each column.

   You should see: age nulls become 50 and the null name becomes unknown. height and bool keep their NULLs.