Initializing the Spark Environment

To begin, a SparkSession must be instantiated with Hive support enabled. This configuration is essential for interacting with Hive metastores and managing partitioned tables effectively.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DataPartitioningJob") \
    .enableHiveSupport() \
    .getOrCreate()

Preparing Source Data

Create a DataFrame containing the records to be ingested. In this example, we generate synthetic data in memory to demonstrate the process without relying on external files.

data_schema = "transaction_id INT, customer_id STRING, amount DOUBLE, transaction_date STRING"
raw_data = [
    (101, "cust_01", 500.00, "2023-11-01"),
    (102, "cust_02", 750.50, "2023-11-01"),
    (103, "cust_03", 120.00, "2023-11-02")
]

input_df = spark.createDataFrame(raw_data, data_schema)

Setting Up the Target Partitioned Table

Before insertion, ensure the target table exists. The following SQL statement creates a partitioned table if it is not already present, partitioning the data by the transaction_date column.

spark.sql("""
CREATE TABLE IF NOT EXISTS daily_transactions (
    transaction_id INT,
    customer_id STRING,
    amount DOUBLE
)
PARTITIONED BY (transaction_date STRING)
STORED AS ORC
""")

Registering a Temporary View and Inserting Data

Register the DataFrame as a temporary view. This allows standard SQL queries to be executed against the in-memory data. Subsequently, an INSERT INTO statement dynamically partitions the data as it writes to the target table.

input_df.createOrReplaceTempView("staging_transactions")

insert_query = """
    INSERT INTO TABLE daily_transactions
    PARTITION (transaction_date)
    SELECT transaction_id, customer_id, amount, transaction_date
    FROM staging_transactions
"""

spark.sql(insert_query)