Standardizing Data Types and Writing Silver Tables

1.Casting bronze strings to real types

Bronze tables usually hold most fields as strings so that unexpected schema changes don't break ingestion. Type casting is one of the operations the silver layer is expected to perform. In PySpark, Column.cast(dataType) takes either a type name string or a DataType instance, and astype is an alias for cast. In SQL, cast(expr AS type) does the same thing, and the :: operator is a synonym for it.

from pyspark.sql.types import StringType
df = spark.createDataFrame(
     [(2, "Alice"), (5, "Bob")], ["age", "name"])
df.select(df.age.cast("string").alias('ages')).collect()

# [Row(ages='2'), Row(ages='5')]

Not every pair of source and target types can be cast. The cast function documents a full matrix of valid combinations. These are the rows that matter most for cleaning bronze:

A valid type pair still doesn't guarantee that every value converts cleanly. Casting a numeric to an integral type truncates toward zero, so cast(5.6 AS INT) returns 5 and cast(-5.6 AS INT) returns -5. Casting a string reads the string as a literal of the target type, which is where malformed bronze values cause trouble.

Date and timestamp strings often need more than a plain cast. Other built-in functions cast between types using format directives: to_date(expr[,fmt]) returns expr cast to a date using an optional format, and to_timestamp(expr[,fmt]) does the same for timestamps. Like cast, they are sensitive to ANSI mode. With an invalid expr or format string they raise an error in ANSI mode and return NULL when ANSI mode is off.

Sources12345

2.When cast fails: errors, ANSI mode and try_cast

> SELECT cast('123' AS INT);
  123

> SELECT cast('123.0' AS INT);
  Error: CAST_INVALID_INPUT

Values that are out of range fail too. cast(128 AS TINYINT) raises CAST_OVERFLOW. This strict behaviour depends on ANSI mode. In Databricks Runtime, if spark.sql.ansi.enabled is false, an overflow wraps instead of raising an error, and a string with an invalid format or invalid characters for the target type becomes NULL. So the CAST_INVALID_INPUT error above is what you see with ANSI mode on. Databricks recommends using an explicit cast or try_cast rather than relying on ANSI_MODE = FALSE. For cleaning, try_cast is usually the right tool. It performs the same cast, but returns NULL where cast would raise an overflow or invalid-format error, and it behaves the same way in both ANSI modes. A row that won't convert then becomes an ordinary null, and the null-handling techniques from bronze cleaning apply to it.

from pyspark.sql.types import LongType
df = spark.createDataFrame(
     [(2, "123"), (5, "Bob"), (3, None)], ["age", "name"])
df.select(df.name.try_cast(LongType())).show()

# +----+
# |name|
# +----+
# | 123|
# |NULL|
# |NULL|
# +----+

PySpark's Column.try_cast was added in Databricks Runtime 15.0. There is one limit to how forgiving try_cast is: it only tolerates bad values. If the type pair itself isn't a valid cast combination, it still raises DATATYPE_MISMATCH.

df = spark.createDataFrame(
     [(2, "123"), (5, "Bob"), (3, None)], ["age", "name"])
df.select(df.name. ? ("double")).show()

Sources156

3.Writing the cleaned result to a silver table

Once the data is clean and typed, it goes to a new silver table. In SQL that is INSERT INTO silver_table SELECT ... FROM bronze_table. When inserting, Databricks casts data types when it can do so safely without losing information. In PySpark, DataFrameWriter.saveAsTable saves a DataFrame as a named table:

spark.createDataFrame([
    (100, "Alice"), (120, "Bob"), (140, "Tom")],
    schema=["age", "name"]
).write.saveAsTable("tblA")

What happens when the target already exists depends on mode. The accepted values are 'append', 'overwrite', 'error' or 'errorifexists' (the default), and 'ignore'. The source documents the behaviour of the first three as follows; it lists 'ignore' as an accepted value but does not describe what it does, so check the documentation before relying on it.

When appending to an existing table, saveAsTable uses column names to find the correct column positions, unlike insertInto.

Sources78

4.Making the silver table enforce what you cleaned

Cleaning code can regress, so the silver table itself can reject bad data on write. Delta Lake applies schema enforcement and constraint checks when data is written. Two constraint types are enforced: NOT NULL and CHECK. Primary and foreign keys are informational only. A CHECK constraint can enforce a value range or a regex pattern (using REGEXP or RLIKE):

CREATE TABLE table_name (
  column_name INT CHECK (column_name >= 0 AND column_name <= 100)
);

NOT NULL blocks new rows that have nulls in the column. You can add it to an existing table only if the column currently contains no nulls. So clean the existing nulls first, then add the constraint.

Sources8

Exam traps

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

1. 1.try_cast never raises an error, whatever types you give it.Why is that wrong?

   try_cast returns NULL for malformed or out-of-range values only. An invalid source and target type combination still raises DATATYPE_MISMATCH.

   Covered in When cast fails: errors, ANSI mode and try_cast

2. 2.Casting an invalid string always raises CAST_INVALID_INPUT.Why is that wrong?

   That is the ANSI-mode behaviour. With spark.sql.ansi.enabled false, a string with an invalid format or invalid characters becomes NULL instead.

   Covered in When cast fails: errors, ANSI mode and try_cast

3. 3.saveAsTable appends to an existing table by default.Why is that wrong?

   The default mode throws an exception when the table exists. Appending requires mode='append'.

   Covered in Writing the cleaned result to a silver table

4. 4.You can add NOT NULL to an existing silver table at any time, and it only affects future rows.Why is that wrong?

   NOT NULL can be enabled on an existing table only if the column has no nulls yet, so existing nulls must be cleaned first.

   Covered in Making the silver table enforce what you cleaned

Practise it for real

Fill nulls, cast a column, write the result to a table and read it back, using only the documented PySpark calls

1. 1.Create the fillna sample: 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"])

   Why: This gives you a small stand-in for a bronze DataFrame, with nulls in several columns.

   You should see: A four-row DataFrame with NULLs in age, height, name and bool.

2. 2.Run filled = df.na.fill({'age': 50, 'name': 'unknown'}), then filled.show()

   Why: A dict fill sets a default for each column. Assigning the result matters, because fillna returns a new DataFrame and leaves df unchanged.

   You should see: age nulls become 50, the null name becomes 'unknown', and height stays NULL.

3. 3.On the filled DataFrame, run filled.select(filled.age.cast("string").alias('ages')).show()

   Why: This practises Column.cast with a type name string, on the column you just filled.

   You should see: An ages column holding string values such as '10', '5' and '50'.

4. 4.Write the cleaned DataFrame with filled.write.saveAsTable("tblA"), then run it again unchanged

   Why: This shows the default mode's behaviour when the table already exists.

   You should see: The first write succeeds. The second throws an exception, because the default mode is 'error'.

5. 5.Run spark.read.table("tblA").sort("age").show(), then spark.sql("DROP TABLE tblA")

   Why: Reading the table back confirms what was written, and dropping it cleans up.

   You should see: The cleaned rows, sorted by age.