!pip install delta-spark
+Traditionally, implementing upsert (update or insert) logic requires separate UPDATE and INSERT statements or complex SQL. This approach can be error-prone and inefficient, especially for large datasets.
+Delta Lake’s merge operation solves this problem by allowing you to specify different actions for matching and non-matching records in a single, declarative statement.
+Here’s an example that demonstrates the power and simplicity of Delta Lake’s merge operation:
+First, let’s set up our initial data:
+# Create sample data for 'customers' DataFrame
+customers_data = [
+ (1, "John Doe", "john@example.com", "2023-01-01 10:00:00"),
+ (2, "Jane Smith", "jane@example.com", "2023-01-02 11:00:00"),
+ (3, "Bob Johnson", "bob@example.com", "2023-01-03 12:00:00"),
+]
+customers = spark.createDataFrame(
+ customers_data, ["customer_id", "name", "email", "last_updated"]
+)
+
+# Create sample data for 'updates' DataFrame
+updates_data = [
+ (2, "Jane Doe", "jane.doe@example.com"), # Existing customer with updates
+ (3, "Bob Johnson", "bob@example.com"), # Existing customer without changes
+ (4, "Alice Brown", "alice@example.com"), # New customer
+]
+updates = spark.createDataFrame(updates_data, ["customer_id", "name", "email"])
+
+# Show the initial data
+print("Initial Customers:")
+customers.show()
+print("Updates:")
+updates.show()
+Initial Customers:
++-----------+-----------+----------------+-------------------+
+|customer_id| name| email| last_updated|
++-----------+-----------+----------------+-------------------+
+| 1| John Doe|john@example.com|2023-01-01 10:00:00|
+| 2| Jane Smith|jane@example.com|2023-01-02 11:00:00|
+| 3|Bob Johnson| bob@example.com|2023-01-03 12:00:00|
++-----------+-----------+----------------+-------------------+
+
+Updates:
++-----------+-----------+--------------------+
+|customer_id| name| email|
++-----------+-----------+--------------------+
+| 2| Jane Doe|jane.doe@example.com|
+| 3|Bob Johnson| bob@example.com|
+| 4|Alice Brown| alice@example.com|
++-----------+-----------+--------------------+
+Next, we create a Delta table from our initial customer data:
+# Define the path where you want to save the Delta table
+delta_table_path = "customers_delta"
+
+# Write the DataFrame as a Delta table
+customers.write.format("delta").mode("overwrite").save(delta_table_path)
+
+# Create a DeltaTable object
+customers_delta = DeltaTable.forPath(spark, delta_table_path)
+
+print("Customers Delta Table created successfully")
+
+Customers Delta Table created successfully
+
+Now, here’s the key part - the merge operation that handles both updates and inserts in a single statement:
+# Assume 'customers_delta' is your target table and 'updates' is your source of new data
+customers_delta.alias("target").merge(
+ updates.alias("source"),
+ "target.customer_id = source.customer_id"
+).whenMatchedUpdate(set={
+ "name": "source.name",
+ "email": "source.email",
+ "last_updated": "current_timestamp()"
+}).whenNotMatchedInsert(values={
+ "customer_id": "source.customer_id",
+ "name": "source.name",
+ "email": "source.email",
+ "last_updated": "current_timestamp()"
+}).execute()
+
+# Verify the updated data
+print("Updated Customers Delta Table:")
+customers_delta.toDF().show()
+Updated Customers Delta Table:
+
++-----------+-----------+--------------------+--------------------+
+|customer_id| name| email| last_updated|
++-----------+-----------+--------------------+--------------------+
+| 2| Jane Doe|jane.doe@example.com|2024-08-20 16:05:...|
+| 3|Bob Johnson| bob@example.com|2024-08-20 16:05:...|
+| 4|Alice Brown| alice@example.com|2024-08-20 16:05:...|
+| 1| John Doe| john@example.com| 2023-01-01 10:00:00|
++-----------+-----------+--------------------+--------------------+
+Appending mismatched data to a Parquet table involves reading the existing data, concatenating it with the new data, and overwriting the existing Parquet file. This approach can be expensive and may lead to schema inconsistencies.
In the following code, the datatype of col3 is supposed to be int64 instead of float64.
Extract features and select only relevant features for each time series.
Learn more about Streaming API in Polars.
As a data scientist, you’re likely familiar with the popular data analysis libraries Pandas and Polars. Both provide powerful tools for working with tabular data, but how do their syntaxes compare?
To begin, we’ll create equivalent dataframes in both Pandas and Polars: