Slowly changing data (SCD) Type 2 operation into Delta tables

customersTable = ... # DeltaTable with schema (customerId, address, current, effectiveDate, endDate)

updatesDF = ... # DataFrame with schema (customerId, address, effectiveDate) # Rows to INSERT new addresses of existing customers newAddressesToInsert = updatesDF \ .alias("updates") \ .join(customersTable.toDF().alias("customers"), "customerid") \ .where("customers.current = true AND updates.address <> customers.address") # Stage the update by unioning two sets of rows # 1. Rows that will be inserted in the whenNotMatched clause # 2. Rows that will either update the current addresses of existing customers or insert the new addresses of new customers stagedUpdates = ( newAddressesToInsert .selectExpr("NULL as mergeKey", "updates.*") # Rows for 1 .union(updatesDF.selectExpr("updates.customerId as mergeKey", "*")) # Rows for 2. ) # Apply SCD Type 2 operation using merge customersTable.alias("customers").merge( stagedUpdates.alias("staged_updates"), "customers.customerId = mergeKey") \ .whenMatchedUpdate( condition = "customers.current = true AND customers.address <> staged_updates.address", set = { # Set current to false and endDate to source's effective date. "current": "false", "endDate": "staged_updates.effectiveDate" } ).whenNotMatchedInsert( values = { "customerid": "staged_updates.customerId", "address": "staged_updates.address", "current": "true", "effectiveDate": "staged_updates.effectiveDate", # Set current to true along with the new address and its effective date. "endDate": "null" } ).execute()

DataLake Vs DeltaLake

Delta Lake is an extension of a data lake that provides ACID (Atomicity, Consistency, Isolation, Durability) transactions, schema enforcement, and data history functionality on top of your existing data lake. In other words, Delta Lake adds reliability and performance optimizations to a data lake.

Here are some key differences:

ACID Transactions:

Data Lake: Traditional data lakes typically lack ACID transactions. They are often designed for scalable storage but may not ensure the consistency and reliability of data transactions.

Delta Lake: Delta Lake, on the other hand, introduces ACID transactions to a data lake environment. This means that operations like inserts, updates, and deletes are atomic and consistent, ensuring the reliability of your data.

Schema Enforcement:

Data Lake: In a regular data lake, there might not be strict enforcement of a schema. Different data files within the lake could have varying structures.

Delta Lake: Delta Lake provides schema enforcement, ensuring that the data conforms to a predefined schema. This helps maintain data consistency and makes it easier to work with structured data.

Data History:

Data Lake: Traditional data lakes might not inherently track changes to the data over time.

Delta Lake: Delta Lake maintains a transaction log that allows you to track changes to your data, enabling features like time travel. This makes it possible to view and revert to previous versions of the data.

Optimizations for Performance:

Data Lake: Regular data lakes might not have optimizations for certain types of queries, leading to potential performance challenges.

Delta Lake: Delta Lake introduces optimizations like data skipping and predicate pushdown to enhance query performance.

In summary, while a data lake is a storage repository that can hold vast amounts of raw data in its native format, Delta Lake is a technology that adds features like ACID transactions, schema enforcement, and data history to make working with data lakes more reliable and efficient. Delta Lake is often used as a way to bring more structure and reliability to the flexible storage capabilities of a data lake.

Predicate Push down in Spark

 Predicate pushdown is a technique in Apache Spark where filters (predicates) are pushed down to the data source level, reducing the amount of data that needs to be processed. This can significantly improve query performance by filtering out unnecessary data early in the data retrieval process.

Let's consider an example with Spark SQL and a DataFrame reading data from a source like Apache Parquet:

from pyspark.sql import SparkSession # Create a Spark session spark = SparkSession.builder.appName("PredicatePushdownExample").getOrCreate() # Assume you have a Parquet file with a schema containing columns: id, name, age, and city parquet_path = "/path/to/your/parquet/data" # Read data from the Parquet file df = spark.read.parquet(parquet_path) # Example of predicate pushdown filtered_df = df.filter(df["age"] > 25) # Show the resulting DataFrame filtered_df.show() # Stop the Spark session spark.stop()

In this example, the filter operation applies a condition (age > 25) on the DataFrame df. When Spark encounters this filter, it can push the filter down to the Parquet data source, meaning that the Parquet reader will only read and process the data that satisfies the condition. This is an optimization because it reduces the amount of data read from disk and processed in-memory.

Behind the scenes, Spark's Catalyst optimizer analyzes the DataFrame operations and tries to push down predicates to the data source whenever possible. The specific capabilities depend on the data source. For example, Parquet supports predicate pushdown, as do many other data sources like ORC and Delta Lake.

Always keep in mind that the effectiveness of predicate pushdown depends on the underlying data source and the type of operations being performed. It's a powerful optimization technique, especially when dealing with large datasets, as it minimizes the amount of data that needs to be processed in a Spark job.

Spark Optimization Techniques

 1. Partitioning:

Ensure that your data is well-partitioned. Adjust the number of partitions using repartition and coalesce to optimize data distribution across nodes.

2. Broadcast hash Join :

Use Broadcast hash join for small tables that can fit in memory. This reduces the need for expensive shuffling in join operations. 

3. Caching and Persistence:

Cache intermediate results or frequently used datasets using persist or cache to avoid recomputation.

4.Memory Tuning:

Adjust the memory configurations for executors and the driver to optimize the memory usage . Balance between executor memory , storage memory and the shuffle memory . 

5.Data Serialization:

Choose efficient serialization formats (e.g., Kryo, Avro) to reduce the size of data transmitted over the network.

6.Shuffling Optimization:

Minimize data shuffling by using operations like reduceByKey instead of groupByKey. Optimize joins, aggregations and group by operations 

Adjust the shuffle partition size using spark.sql.shuffle.partitions.

7.Dynamic Allocation:

Enable dynamic allocation (spark.dynamicAllocation.enabled) to adjust the number of executors dynamically based on workload.This can help in resource utilization. 

8.Code Optimization:

Write efficient Spark transformations and actions. Minimize the use of unnecessary collect and repartition operations.Minimize the use of custom UDFs and prefer built-in Spark functions when possible.

9.Hardware Considerations:

Properly provision hardware resources , considering factors like CPU , memory to match the workload and data processing requirement .

10. Column Pruning:

When reading data only select the columns needed for processing to reduce  I/O and memory requirements .

11.Predicate Pushdown :

Predicate pushdown is a technique in Apache Spark where filters (predicates) are pushed down to the data source level, reducing the amount of data that needs to be processed. This can significantly improve query performance by filtering out unnecessary data early in the data retrieval process.

# Example of predicate pushdown

filtered_df = df.filter(df["age"] > 25)

In this example, the filter operation applies a condition (age > 25) on the DataFrame df. When Spark encounters this filter, it can push the filter down to the Parquet data source, meaning that the Parquet reader will only read and process the data that satisfies the condition. This is an optimization because it reduces the amount of data read from disk and processed in-memory.

12.Parquet file Format:

Choose efficient file formats like Parquet, which provides columnar storaand can significantly reduce I/O and query performance . 

13.Task Paralleism:

Increase the level of parallelism by optimiziing the number of tasks. This involves adjusting the number of partitions and degree of paralleism in operations.

14. Monitor and Tune Spark UI:

Monitor the jobs and identify the bottlenecks. Tune configuration based on the insight gained from monitoring.

Python to Mysql Connectivity

 import mysql.connector

connection = mysql.connector.connect( host="localhost", user="yourusername", password="yourpassword", database="mydatabase" ) mycursor = connection.cursor() mycursor.execute("SELECT name, address FROM customers") myresult = mycursor.fetchall() for x in myresult: print(x)

Count words in a file in Python program

 with open ('E:\\source_files\\file1.txt','r') as file:

    #Read the entire file
 content = file.read()
print(content)

#C:\bigdata\anaconda3\envs\py38\python.exe C:\bigdata\PycharmProjects\Test_Project\test1.py

#Bangalore is Karnataka
#Chennai is Tamilnadu

#split the content into words
words = content.split()
print(f'output after split is  :{words}')
# output after split is  :['Bangalore', 'is', 'Karnataka', 'Chennai', 'is', 'Tamilnadu']

#count the number of words:
word_count = len(words)
print(f'final output count  is  :{word_count}')
#final output count  is  :6

#1st Simple Glue Job

 #1st Simple Glue Job

import sys from awsglue.transforms import * from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from pyspark.sql.functions import * from awsglue.context import GlueContext from awsglue.job import Job ## @params: [JOB_NAME] args = getResolvedOptions(sys.argv, ['JOB_NAME']) sc = SparkContext() glueContext = GlueContext(sc) spark = glueContext.spark_session data ="s3://rishita2023poc/input/bank-full.csv" # \ Slash symobol should be given after Parenthesis closure bracket but before '.' symobol. adf= spark.read.format ("csv")\ .option("header","true")\ .option("inferSchema","true")\ .option("sep",";")\ .load(data) adf.show() res=adf.where(col("age")>60) op="s3://rishita2023poc/output" res.write.format("csv") \ .option("header","true")\ .mode("overwrite")\ .save(op) ''' ### To Store the output to MYSQL RDS db ,add the dependency jar file from S3 Bucket path in dependency section. #Dependent Jars path: s3://rishita2023poc/drivers/mysql-connector-java-8.0.12.jar host="jdbc:mysql://mysqldb.cwkqaojgxfrd.ap-south-1.rds.amazonaws.com:3306/newdb" res.write.format("jdbc")\ .option("url",host)\ .option("user","myuser")\ .option("password","mypassword")\ .option("driver","com.mysql.cj.jdbc.Driver")\ .option("dbtable","gluebank60")\ .save() ''' job = Job(glueContext) job.init(args['JOB_NAME'], args) job.commit()