Example Project: Comprehensive ETL Pipeline in PySpark

---

Let’s create a comprehensive example project that demonstrates these operations.

1. Setup Spark Session
2. Load Data from CSV, Hive, Oracle, and create DataFrames
3. Apply Transformations
4. Perform Actions
5. Persist DataFrames
6. Convert to Pandas for further analysis and visualization

1. Setup Spark Session

from pyspark.sql import SparkSession

# Initialize Spark session
spark = SparkSession.builder 
    .appName("ETL Pipeline Example") 
    .enableHiveSupport() 
    .getOrCreate()

2. Load Data

From CSV

df_csv = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

From Hive

df_hive = spark.sql("SELECT * FROM database.table_name")

From Oracle

jdbc_url = "jdbc:oracle:thin:@hostname:port:SID"
connection_properties = {
    "user": "username",
    "password": "password",
    "driver": "oracle.jdbc.driver.OracleDriver"
}

df_oracle = spark.read.jdbc(jdbc_url, "schema.table_name", properties=connection_properties)

3. Apply Transformations

# Select columns
df_selected = df_csv.select("name", "age", "salary")

# Filter rows
df_filtered = df_selected.filter(df_selected["age"] > 30)

# Group by and aggregate
df_grouped = df_filtered.groupBy("department").mean("salary")

# Join with another DataFrame
df_joined = df_grouped.join(df_hive, df_grouped["department"] == df_hive["dept_id"])

# Add new column
df_with_new_col = df_joined.withColumn("new_col", df_joined["salary"] + 1000)

# Drop column
df_final = df_with_new_col.drop("dept_id")

4. Perform Actions

# Show DataFrame
df_final.show()

# Collect DataFrame
rows = df_final.collect()
for row in rows:
    print(row)

# Count rows
print(df_final.count())

# Describe DataFrame
df_final.describe().show()

5. Persist DataFrames

# Cache DataFrame
df_final.cache()

# Persist DataFrame
from pyspark import StorageLevel
df_final.persist(StorageLevel.MEMORY_AND_DISK)

6. Convert to Pandas for Further Analysis and Visualization

import pandas as pd

# Convert to Pandas DataFrame
pdf = df_final.toPandas()

# Analyze with Pandas
print(pdf.describe())

# Visualize with Matplotlib
import matplotlib.pyplot as plt

pdf['salary'].hist()
plt.show()

you can use SQL expressions inside DataFrame operations,

including the filter or where functions, in PySpark. This is done

Date and Time Functions- Pyspark Dataframes

https://www.hintstoday.com/pyspark-dataframe-programming-operations-functions-all-statements-syntax-with-examples/date-and-time-functions-pyspark-dataframes

String Manipulation on PySpark DataFrames

https://www.hintstoday.com/pyspark-dataframe-programming-operations-functions-all-statements-syntax-with-examples/string-manipulation-on-pyspark-dataframes

Window functions in PySpark on Dataframes

https://www.hintstoday.com/pyspark-dataframe-programming-operations-functions-all-statements-syntax-with-examples/window-functions-in-pyspark-on-dataframe

---

Head to Next

Pages: 1 2 3 4 5 6 7 8 9 10 11 12