UPSERTS and DELETES using AWS Glue and Delta Lake
Tags: aws, tutorial, bigdata, analytics
Published: 2021-07-21 03:45:42

The purpose of this blog post is to demonstrate how you can enable your Data Lake to be ACID-compliant, that is, having the same functionality as a database. This will allow you to do UPSERTS and DELETES directly to your data lake

Let me start first by defining what a Data Lake is:

From [AWS] (https://aws.amazon.com/big-data/datalakes-and-analytics/what-is-a-data-lake/)

> A data lake is a __centralized repository__ that allows you to store all your structured and unstructured data at any scale. You can __store your data as-is__, without having to first structure the data, and run different types of analytics—from dashboards and visualizations to __big data processing__, real-time analytics, and machine learning to guide better decisions.

Data Lake
---

A data lake is scalable, performant, secure, and cost-efficient. And it has played a crucial part of an organization's Data Analytics pipeline. So what's the problem?

Well, __updates__.

We all know that data lakes are [immutable] (https://www.ctl.io/developers/blog/post/immutability) - _the idea that data or objects should not be modified after they are created_; how do we then go beyond that immutability? 

Delta Lake
--- 

The answer is [Delta Lake] (https://delta.io/).

> An [open-source] (https://github.com/delta-io/delta) storage layer that brings scalable, ACID transactions to Apache Spark™ and big data workloads. It provides serializability, the strongest level of isolation level. Scalable Metadata Handling, Time Travel, and is 100% compatible with Apache Spark APIs.


Basically, it allows you to do DELETES and UPSERTS directly to your data lake. 

How Spark Fails ACID
---

We all know our beloved Spark doesn't support ACID transactions, but to be fair, it isn't really built to address that kind of specific use case.

I came across a blog post from [kundankumarr] (https://blog.knoldus.com/spark-acid-compliant-or-not/), explaining how Spark fails ACID. 

__A__tomicity & __C__onsistency
> __Atomicity__ states that it should either write full data or nothing to the data source when using spark data frame writer. __Consistency__, on the other hand, ensures that the data is always in a valid state.

__I__solation & __D__urability
> We know that when a transaction is in process and not yet committed, it must remain isolated from any other transaction. This is called __Isolation__ Property. It means writing to a data set shouldn’t impact another concurrent read/write on the same data set.

> Finally, __Durability__. It is the ACID property which guarantees that transactions that have committed will survive permanently. However, when Spark doesn’t correctly implement the commit, then all the durability features offered by the storage goes for a toss.


AWS Glue and Delta Lake
---

This part demonstrates how you can use Delta Lake with AWS Glue. 

These are the services that will be used in this exercise:
### AWS Glue

> a serverless data integration service that makes it easy to discover, prepare, and combine data for analytics, machine learning, and application development.

### Amazon Athena

> an interactive query service that makes it easy to analyze data in Amazon S3 using standard SQL. Athena is serverless, so there is no infrastructure to manage, and you pay only for the queries that you run.

### Amazon S3

> an object storage service that offers industry-leading scalability, data availability, security, and performance.


This is what we'll be doing:

![process-flow](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/qr0m67akjycfnyrffza5.PNG)

Basically, I have an `initial data` then I want to apply changes to the `Sales` and `Profit` column. Then the table in the __AWS Glue Data Catalog__ should be able to capture that changes. Just a basic update to the data.

So, let's start!

### Pre-requisites
First, download the data [here] (https://www.kaggle.com/bravehart101/sample-supermarket-dataset) - I used Tableau's Superstore Dataset, this one is on Kaggle, you may need to register for an account to download.

Then, you need to download the Delta Lake `.jar` file to access it's libraries. You can download it [here] (https://search.maven.org/artifact/io.delta/delta-core_2.11/0.6.1/jar). Upload it on your S3 Bucket and take note of the S3 path, we'll use this as a reference later. 

❗ _As of this writing, Glue's Spark Engine (v2.4) only supports v0.6.1 of Delta Lake since versions beyond that were implemented in Spark 3.0._

❗❗❗ _UPDATE: AWS GLUE 3.0 WAS RELEASED ON AUGUST 2021! Check out my blog post on this one:_ ❗❗❗

{% link https://dev.to/awscommunity-asean/sql-based-inserts-deletes-and-upserts-in-s3-using-aws-glue-3-0-and-delta-lake-42f0 %}

### AWS Glue

Navigate to AWS Glue then proceed to the creation of an ETL Job. Specify the `This job runs` to `A new script to be authored by you`. This will allow you to have a custom spark code.

Under `Security configuration, script libraries, and job parameters (optional)`, specify the location of where you stored the `.jar` file as shown below:
![etl-job](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/lab1rujbugkn6wmo7ltl.PNG)

Then on blank script page, paste the following code:

```python
from delta import *
from pyspark.sql.session import SparkSession
```
This imports the SparkSession libraries as well as the Delta Lake libraries.

```python
# Initialize Spark Session with Delta Lake
spark = SparkSession \
  .builder \
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
  .getOrCreate()
```

This code initializes the `SparkSession` along with the Delta Lake configurations. 

```python
# Read Source
inputDF = spark.read.format("csv") \
.option("header", "true") \
.load('s3://delta-lake-ia-test/raw/')
```
We read the source csv file as a Spark DataFrame.

```python
# Write data as DELTA TABLE
inputDF.write.format("delta") \
.mode("overwrite") \
.save("s3a://delta-lake-ia-test/current/")
```

Then we output it as a `Delta` format. 

❗ Notice the use of `s3a` prefix in the save path, it is essential to use the `s3a` prefix instead of the standard `s3` as the path. As using the `s3` prefix, will throw an `UnsupportedFileSystemException` error. Followed by a `fs.AbstractFileSystem.s3.impl=null: No AbstractFileSystem configured for scheme: s3`.

More on the differences of `s3` and `s3a` [here] (https://stackoverflow.com/questions/33356041/technically-what-is-the-difference-between-s3n-s3a-and-s3)

```python
# Generate MANIFEST file for Athena/Catalog
deltaTable = DeltaTable.forPath(spark, "s3a://delta-lake-ia-test/current/")
deltaTable.generate("symlink_format_manifest")

```

Athena supports reading from external tables using a `manifest` file, which is a text file containing the list of data files to read for querying a table. Running the above code will generate a `manifest` file.

Read more about Delta Lake's integration for Presto and Athena [here] (https://docs.delta.io/0.6.1/presto-integration.html)


Final Code:

```python
from delta import *
from pyspark.sql.session import SparkSession

spark = SparkSession \
  .builder \
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
  .getOrCreate()


# Read Source
inputDF = spark.read.format("csv").option("header", "true").load('s3://delta-lake-ia-test/raw/')

# Write data as DELTA TABLE
inputDF.write.format("delta").mode("overwrite").save("s3a://delta-lake-ia-test/current/")

# Generate MANIFEST file for Athena/Catalog
deltaTable = DeltaTable.forPath(spark, "s3a://delta-lake-ia-test/current/")
deltaTable.generate("symlink_format_manifest")

```

### Amazon Athena

In your S3 bucket, you should see a `_symlink_format_manifest` prefix/folder. This will be used by Amazon Athena for mapping out the parquet files.

Create your table using the code below as a reference:
```sql
CREATE EXTERNAL TABLE IF NOT EXISTS "default"."superstore" ( 
row_id STRING,
order_id STRING,
order_date STRING,
ship_date STRING,
ship_mode STRING,
customer_id STRING,
customer_name STRING,
segment STRING,
country STRING,
city STRING,
state STRING,
postal_code STRING,
region STRING,
product_id STRING,
category STRING,
sub_category STRING,
product_name STRING,
sales STRING,
quantity STRING,
discount STRING,
profit STRING
) ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 

STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://delta-lake-ia-test/current/_symlink_format_manifest/'

```

~~I'm lazy.~~ I've made things simple by using `STRING` for all columns.

Note that you have to define the __table name__ as it is when you  you wrote it as a __delta table__. Or else you'll get blank results when querying with Athena.

```sql
-- Run a simple select
SELECT *
FROM "default"."superstore"
LIMIT 10
```
![athena-sample-resultset](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ypbn4wr5tmr2ibiztepx.PNG)


Recap:

* Read from a CSV
* Created a Spark DataFrame from the CSV
* Written the DataFrame as a Delta Table
* Made a manifest file
* Created an external table in Athena
* Query sample data


What we'll do next:

* Read the updates from the CSV
* Make an update based on the new files
* Generate/update the manifest file


Let's add another Glue ETL Job for the updates.

I have manually modified my raw data to simulate the updates, I just plug in the `99999` values in the sales and profit for the first 15 rows. Feel free to have your own modifications. 

After which, upload it to your S3 Bucket in a different location. 

```python
from delta import *
from pyspark.sql.session import SparkSession


spark = SparkSession \
  .builder \
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
  .getOrCreate()

```

Nothing new here.


```python
# Read updates
df_updates = spark.read.format("csv").option("header", "true").load('s3://delta-lake-ia-test/updates/')

# Read current as DELTA TABLE
df = DeltaTable.forPath(spark, "s3a://delta-lake-ia-test/current/")
``` 
First line is a typical read from csv code.

The next line creates a `DeltaTable` object, which allows us to call functions in the delta package.

```python
# UPSERT process
final_df = df.alias("full_df").merge(
    source = df_updates.alias("append_df"),
    condition = expr("append_df.row_id = full_df.row_id"))\
  .whenMatchedUpdateAll() \
  .whenNotMatchedInsertAll() \
  .execute()
```
One of this is the `merge(source, condition)` function, which:
> Merges the data from the _source_ DataFrame based on the given merge _condition_.

First, we take a `DeltaTable DataFrame` object, then give it an alias. We then call the `merge()` function, supplying the `Parameters` with the our `Arguments`. Which, in this case, is the updates `DataFrame` and the merge condition.

Then, we call the `whenMatchedUpdateAll(condition=None)`
> Updates all the columns of the matched table row with the values of the corresponding columns in the source row. If a `condition` is specified, then it must be true for the new row to be updated.

to have the code update all the columns.

If the condition specified in the `merge()` function doesn't match, then we do a `whenNotMatchedInsertAll(condition=None)`
> Insert a new target Delta table row by assigning the target columns to the values of the corresponding columns in the source row. If a `condition` is specified, then it must evaluate to true for the new row to be inserted.

Lastly, we call the `execute()` function to sum it up
> Execute the merge operation based on the built matched and not matched actions.

```python
# Generate new MANIFEST file
final_df = DeltaTable.forPath(spark, "s3a://delta-lake-ia-test/current/")
final_df.generate("symlink_format_manifest")
```

Then we update the `manifest` file.

For more functions in the library, kindly refer to the [official docs] (https://docs.delta.io/0.6.1/api/python/index.html)


Final code:

```python

from delta import *
from pyspark.sql.session import SparkSession


spark = SparkSession \
  .builder \
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
  .getOrCreate()


# Read updates
df_updates = spark.read.format("csv").option("header", "true").load('s3://delta-lake-ia-test/updates/')

# Read current as DELTA TABLE
df = DeltaTable.forPath(spark, "s3a://delta-lake-ia-test/current/")

# UPSERT process
final_df = df.alias("full_df").merge(
    source = df_updates.alias("append_df"),
    condition = expr("append_df.row_id = full_df.row_id"))\
  .whenMatchedUpdateAll() \
  .whenNotMatchedInsertAll() \
  .execute()

# Generate new MANIFEST file
final_df = DeltaTable.forPath(spark, "s3a://delta-lake-ia-test/current/")
final_df.generate("symlink_format_manifest")
```


Now, try querying your updated in Athena. It should show the most updated data. 

![athena-updated-dataset](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/y7fljgeli4qrnznai2ij.PNG)

---

Conclusion and Final Thoughts
--- 

This blog post demonstrated how you can leverage ACID transactions in your data lake.

Having a functionality like this is helpful especially if you have requirements such as Change Data Capture (CDC). Curious to know how you guys implemented such things. Let me know in the comments! 

I've read a couple of articles and blog posts whether a Data Lake should be `immutable` or not.

From _O'Reilly_, [Data Lake for Enterprises by Tomcy John, Pankaj Misra] (https://www.oreilly.com/library/view/data-lake-for/9781787281349/91414449-b6a4-463a-bdf8-ac578047e7ff.xhtml) on the topic of __Immutable Data Principle__
> The data should be stored in a raw format from the different source systems. More importantly, the data stored should be immutable in nature. 
> 
> By making it immutable, it inherently takes care of human fault tolerance to at least some extent and takes away errors with regards to data loss and corruption. It allows data to be selected, inserted, and not updated or deleted. 
> 
> To cater to fundamental fast processing/performance, the data is usually stored in a denormalized fashion. Data being immutable makes the system in general simpler and more manageable.

From [SQLServerCentral] (https://www.sqlservercentral.com/editorials/should-the-data-lake-be-immutable) by Steve Jones on the topic of _Should the Data Lake be Immutable?_

> Imagine I had a large set of data, say GBs in a file, would I want to download this and change a few values before uploading it again? Do we want a large ETL load process to repeat? 
> 
> Could we repeat the process and reload a file again? I don't think so, but it's hard to decide. After all, the lake isn't the source of data; that is some other system.
> 
> Maybe that's the simplest solution, and one that reduces complexity, downtime, or anything else that might be involved with locking and changing a file.

Comments on the topic can be found [here](https://www.sqlservercentral.com/forums/topic/should-the-data-lake-be-immutable).

There is a comment from roger.plowman

> I suspect immutability should be asked after asking if you should even have the data lake or warehouse in the first place.

What do you guys think? Would love to hear your thoughts!


Speaking of Data Lakes vs Data Warehouse, there's also this very interesting concept I picked up from one of the AWS Community Builder ([Joel Farvault] (/jol_farvault_72301b8e349)), it is called [Data Mesh Architecture] (https://martinfowler.com/articles/data-monolith-to-mesh.html), in which they describe it as the _next enterprise data platform architecture_. I'll leave it up for you to read on about. AWS also made a [blog post] (https://aws.amazon.com/blogs/big-data/design-a-data-mesh-architecture-using-aws-lake-formation-and-aws-glue/) on this using AWS Lake Formation.


There is also a youtube video back from the AWS DevDay Data & Analytics conducted on July 14, 2021. Where AWS Technical Evangelists; [Javier Ramirez] (https://aws.amazon.com/developer/community/evangelists/javier-ramirez/) and [Ricardo Sueiras] (https://aws.amazon.com/developer/community/evangelists/ricardo-sueiras/) discusses the [Data Mesh Architecture] (https://youtu.be/l3sV9YxIcSo?t=5558).

Additionally, here is a video about [Data Mesh in practice] (https://www.youtube.com/watch?v=eiUhV56uVUc&t=3s) for Europe's biggest online fashion retailer. 


Hope this helps! Let me know if you have questions below.

---

Happy coding!


_P.S. CloudFormation stack is on going_ ⚙