Hello, Habr! I present to your attention the translation of the article. by Burak Yavuz, Brenner Heintz, and Denny Lee, prepared ahead of the course launch from OTUS.
Data, much like our experience, continuously accumulates and evolves. To keep pace, our mental models of the world must adapt to new data, some of which introduces new dimensions—new ways to observe things previously unknown to us. These mental models are little different from table schemas that define how we classify and process new information.
This brings us to the topic of schema management. As business tasks and requirements change over time, so does the structure of your data. Delta Lake makes it easy to implement new dimensions as data changes. Users have access to a straightforward semantics for managing their table schemas. These tools include Schema Enforcement, which protects users from inadvertently cluttering their tables with errors or unnecessary data, and Schema Evolution, which allows for the automatic addition of new columns with valuable data in the appropriate places. In this article, we will delve into the use of these tools.
Understanding Table Schemas
Every DataFrame in Apache Spark contains a schema that defines the shape of the data, such as data types, columns, and metadata. With Delta Lake, the table schema is stored in JSON format within the transaction log.
What is Schema Enforcement?
Schema Enforcement, also known as Schema Validation, is a protective mechanism in Delta Lake that ensures data quality by rejecting records that do not conform to the table schema. Like a hostess at the registration desk of a popular restaurant who only accepts reservations, it checks whether each column of data being entered into the table is on the corresponding list of expected columns (in other words, whether each has a 'reservation'), and rejects any records with columns not on the list.
How does Schema Enforcement work?
Delta Lake uses schema enforcement upon writing, meaning that all new records being added to the table are checked for compatibility with the schema of the target table at write time. If the schema is incompatible, Delta Lake fully rolls back the transaction (data is not written) and raises an exception to alert the user about the mismatch.
To determine the compatibility of a record, Delta Lake uses the following rules. The DataFrame being written:
- must not contain additional columns that are not present in the target table schema. Conversely, it is acceptable if the incoming data does not contain all the columns from the table—those columns will simply be assigned null values.
- must not have data types for its columns that differ from those of the columns in the target table. For instance, if a column in the target table contains data of StringType, but the corresponding column in the DataFrame contains data of IntegerType, forcing schema enforcement will raise an exception and prevent the write operation from proceeding.
- must not include column names that differ only in case. This means you cannot have columns named 'Foo' and 'foo' defined in the same table. While Spark can operate in case-sensitive or case-insensitive mode (by default) when using it, Delta Lake preserves case but remains case-insensitive within schema storage. Parquet is case-sensitive when storing and retrieving column information. To avoid potential errors, data corruption, or loss (which we have personally encountered with Databricks), we decided to implement this restriction.
To illustrate this, let’s take a look at what happens in the code example below when attempting to add some newly generated columns to a Delta Lake table that is not yet configured to accept them.
# Сгенерируем DataFrame ссуд, который мы добавим в нашу таблицу Delta Lake
loans = sql("""
SELECT addr_state, CAST(rand(10)*count as bigint) AS count,
CAST(rand(10) * 10000 * count AS double) AS amount
FROM loan_by_state_delta
""")
# Вывести исходную схему DataFrame
original_loans.printSchema()
root
|-- addr_state: string (nullable = true)
|-- count: integer (nullable = true)
# Вывести новую схему DataFrame
loans.printSchema()
root
|-- addr_state: string (nullable = true)
|-- count: integer (nullable = true)
|-- amount: double (nullable = true) # new column
# Попытка добавить новый DataFrame (с новым столбцом) в существующую таблицу
loans.write.format("delta")
.mode("append")
.save(DELTALAKE_PATH)
Returns:
A schema mismatch detected when writing to the Delta table.
To enable schema migration, please set:
'.option("mergeSchema", "true")'
Table schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
Data schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
-- amount: double (nullable = true)
If Table ACLs are enabled, these options will be ignored. Please use the ALTER TABLE command for changing the schema.Instead of automatically adding new columns, Delta Lake enforces the schema and halts the write operation. To assist in identifying which column (or columns) is causing the mismatch, Spark outputs both schemas from the stack trace for comparison.
What is the benefit of enforcing schema?
Since schema enforcement represents a fairly strict check, it serves as an excellent tool for acting as a gatekeeper of clean, fully transformed datasets ready for production or consumption. It is typically applied to tables that directly serve data:
- Machine learning algorithms
- BI dashboards
- Data analytics and visualization tools
- Any production system requiring strictly structured, strongly typed semantic schemas.
To prepare your data for this final barrier, many users employ a simple multi-hop architecture that gradually adds structure to their tables. To learn more about this, you can check out the article
Of course, schema enforcement can be used anywhere in your pipeline, but be aware that streaming writes to a table can be frustrating if, for example, you've forgotten that you've added another column to the incoming data.
Preventing data dilution
At this point, you might be wondering what all the fuss is about. After all, an unexpected 'schema mismatch' error can trip you up in your workflow, especially if you are new to Delta Lake. Why not just allow the schema to change as needed so I can write my DataFrame, come what may?
As the old saying goes, 'an ounce of prevention is worth a pound of cure.' At some point, if you don't take care to enforce your schema, data type compatibility issues will raise their ugly heads—seemingly homogeneous sources of raw data may contain edge cases, corrupted columns, poorly formed mappings, or other scary things that haunt your nightmares. The best approach is to stop these foes at the gate—with schema enforcement—and deal with them in the light, rather than later, when they start prowling in the dark depths of your working code.
Enforcing a schema ensures that your table's schema won't change unless you explicitly allow it. This prevents data 'dilution' that can occur when new columns are added so frequently that previously valuable, compact tables lose their significance and usability due to data flooding. By encouraging you to be deliberate, set high standards, and expect quality, schema enforcement does exactly what it's intended to do — helps you remain diligent while keeping your tables clean.
If upon further consideration you decide that you actually we need need to add a new column — no problem, here’s a one-liner fix. The solution is schema evolution!
What is schema evolution?
Schema evolution is a feature that allows users to easily modify the current schema of a table according to data that changes over time. It is most often used when performing an add or overwrite operation to automatically adapt the schema to include one or more new columns.
How does schema evolution work?
Following the example in the previous section, developers can easily use schema evolution to add new columns that were previously rejected due to schema mismatches. Schema evolution is activated by adding .option('mergeSchema', 'true') to your Spark command .write or .writeStream.
# Добавьте параметр mergeSchema
loans.write.format("delta")
.option("mergeSchema", "true")
.mode("append")
.save(DELTALAKE_SILVER_PATH)To view the graph, execute the following Spark SQL query
# Создайте график с новым столбцом, чтобы подтвердить, что запись прошла успешно
%sql
SELECT addr_state, sum(`amount`) AS amount
FROM loan_by_state_delta
GROUP BY addr_state
ORDER BY sum(`amount`)
DESC LIMIT 10 
Alternatively, you can set this option for the entire Spark session by adding spark.databricks.delta.schema.autoMerge = True to the Spark configuration. But use this cautiously, as schema enforcement will no longer warn you about unintended schema mismatches.
By including the parameter mergeSchema, all columns that are present in the DataFrame but absent in the target table are automatically added to the end of the schema within the write transaction. Nested fields may also be added and will be included at the end of the corresponding column structures.
Data engineers and scientists can use this option to add new columns (possibly a newly tracked metric or this month's sales metric column) to their existing machine learning production tables without disrupting existing models based on old columns.
The following types of schema changes are permissible as part of the schema evolution when adding or overwriting a table:
- Adding new columns (this is the most common scenario)
- Changing data types from NullType -> any other type or promotion from ByteType -> ShortType -> IntegerType
Other changes not permitted within schema evolution require the schema and data to be overwritten by additions .option("overwriteSchema", "true"). For example, if the column "Foo" was originally an integer and the new schema was to be a string data type, then all Parquet files (data) would need to be rewritten. Such changes include:
- removing a column
- changing the data type of an existing column (in place)
- renaming columns that differ only in case (for example, "Foo" and "foo")
Finally, with the upcoming release of Spark 3.0, explicit DDL (using ALTER TABLE) will be fully supported, allowing users to perform the following operations on table schemas:
- adding columns
- changing column comments
- setting table properties that define table behavior, such as setting transaction log retention duration.
What are the benefits of schema evolution?
Schema evolution can be utilized whenever you intend to change the schema of your table (as opposed to cases where you accidentally added columns to your DataFrame that should not be there). This is the easiest way to migrate your schema because it automatically adds the correct column names and data types without needing to explicitly declare them.
Conclusion
Enforcement of schema applies to any new columns or other schema changes that are not compatible with your table. By setting and maintaining these high standards, analysts and engineers can rely on their data having the highest level of integrity, reasoning about it clearly and explicitly, which enables them to make more effective business decisions.
On the other hand, schema evolution complements schema enforcement, simplifying the implied automatic changes to the schema. Ultimately, it shouldn't be complicated — just adding a column.
Schema enforcement is the yin, while schema evolution is the yang. Together, these features effectively simplify noise suppression and signal tuning.
We would also like to thank Mukul Murti and Pranav Anand for their contributions to this article.
Other articles in this series:
Related articles
Source: habr.com
