spark-delta.md

Spark and Delta

• Delta Basics
• Errors
  • Cannot create spark delta table from JSON input
• Partitioning with delta

Delta Basics

Compared to parquet:

• the file format is the same
• but there is an additional _delta_log directory
• this contains a log of events
• it contains info on the schema
• it allows 'time travel' to view data at a point in time

While some ingests will still read the data (eg. basic pandas reading data from a path) it will not take into account the latest version of the data. It will read all the parquet files as if all the data were valid; resulting in a mish-mash of old/new/dupliated data.

Errors

Cannot create spark delta table from JSON input

Error in SQL statement: AnalysisException: Cannot create table ('`....`'). 
The associated location ('s3://...') is not empty but it's not a Delta table

Error in SQL statement: AnalysisException: REPLACE TABLE is only supported with v2 tables.

py4j.protocol.Py4JJavaError: An error occurred while calling o24.load.
: java.lang.ClassNotFoundException: Failed to find data source: delta.

You can just install delta-spark PyPi package using pip install delta-spark (it will pull pyspark as well), and then refer to it.

Or you can add a configuration option that will fetch Delta package.

It's .config("spark.jars.packages", "io.delta:delta-core_2.12:"). For Spark 3.1 Delta versions is 1.0.0 (see releases mapping docs for more information).

Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: Provided Maven Coordinates must be in the form 'groupId:artifactId:version'. The coordinate provided is: /opt/processes/jars/delta-core_2.12-1.0.0.jar

not .config('spark.jars.packages', f'{os.environ["some_env_var"]}/delta-core_2.12-1.0.0.jar')

instead use .config('spark.jars.packages', 'io.delta:delta-core_2.12-1.0.0.jar')

  delta-core_2.12
  
  this will auto download file find it
  find ~ -name delta-core_2.12-1.0.0.jar
  
  and add to your own jars dir

  : org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "s3"

use .config('fs.s3a.aws.credentials.provider', 'com.amazonaws.auth.DefaultAWSCredentialsProviderChain')

  py4j.protocol.Py4JJavaError: An error occurred while calling o35.load.
: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

 WARNING: loading partitions directly with delta is not recommended.
If you are trying to read a specific partition, use a where predicate.

CORRECT: spark.read.format("delta").load("/data").where("part=1")

INCORRECT: spark.read.format("delta").load("/data/part=1")

---

Partitioning with delta

Schema evolution

When you run CREATE TABLE with a LOCATION that already contains data stored using Delta Lake, Delta Lake does the following:

If you specify only the table name and location, for example:

SQL Copy to clipboardCopy CREATE TABLE default.people10m USING DELTA LOCATION '/tmp/delta/people10m' the table in the metastore automatically inherits the schema, partitioning, and table properties of the existing data. This functionality can be used to “import” data into the metastore.

https://docs.databricks.com/delta/delta-update.html#automatic-schema-evolution

• Schema evolution occurs only when there is either an updateAll (UPDATE SET *) or an insertAll (INSERT *) action, or both.
• In Databricks Runtime 7.4 and below, merge supports schema evolution of only top-level columns, and not of nested columns. (running DBR 10+)