# Streaming Deduplication and Quality Enforcement

This program demonstrates a **real-time data pipeline** using Spark Structured Streaming to handle **deduplication** and **data quality enforcement** on streaming data from CSV files.

# **Objective**

The program achieves the following:

1. **Ingest data** from a directory containing CSV files.
    
2. **Deduplicate records** based on unique identifiers (`event_id`) and keep the latest record within a specified time window.
    
3. **Enforce data quality rules** by filtering and handling invalid or missing data.
    
4. **Output processed data** to the console for debugging and verification.
    

# Code Walkthrough

## Reading Streams

#### **Spark Session Initialization**

The program starts by initializing a **SparkSession**, which is the entry point for Spark Structured Streaming applications.

```python
spark = SparkSession.builder \
    .appName("Streaming Deduplication and Quality Enforcement") \
    .master("local") \
    .getOrCreate()
```

#### **Schema Definition**

A custom schema is defined to specify the structure of the incoming CSV files:

```python
CUSTOM_SCHEMA = StructType([
    StructField("event_id", StringType(), True),    # Unique event identifier
    StructField("name", StringType(), True),       # Name of the entity
    StructField("timestamp", TimestampType(), True),  # Event timestamp
    StructField("value", IntegerType(), True)      # Numeric value for the event
])
```

#### **Reading Streaming Data**

The program reads streaming data from a folder (`../resources/dataset/data_quality/input`) using the CSV format:

```python
streaming_df = spark.readStream \
    .format("csv") \
    .schema(CUSTOM_SCHEMA) \
    .load("../resources/dataset/data_quality/input")
```

## **Deduplication Using Time-Based Windowing**

Deduplication is achieved by grouping records using a **time-based window** and `event_id`. Records with the latest `timestamp` are kept.

**Add Watermark and Grouping**

```python
streaming_df.withWatermark("timestamp", "10 minutes").groupBy(
    window(col("timestamp"), "10 minutes"),  # Group by 10-minute time windows
    col("event_id")                         # Group by event_id
)
```

**Aggregate Functions**

```python
.agg(
    max("timestamp").alias("latest_timestamp"),  # Latest timestamp per group
    max("value").alias("value"),                # Latest value per group
    first("name").alias("name")                 # First non-null name
)
```

## Data Quality Enforcement

Data quality is enforced by applying filters and handling null values:

```python
quality_enforced_df = deduplicated_df \
    .filter((col("value").isNotNull()) & (col("value") > 0)) \
    .fillna({"name": "Unknown"})  # Replace null names
```

**Writing to the Console**

The processed data is written to the console using the `writeStream` method:

```python
quality_enforced_df.writeStream \
    .format("console") \
    .outputMode("complete") \
    .start() \
    .awaitTermination()
```

## Input Data

**Input**

```python
event_id,name,timestamp,value
1,John,2024-12-10 10:00:00,50
1,John,2024-12-10 10:05:00,60
2,Alice,2024-12-10 10:03:00,30
3,,2024-12-10 10:10:00,-10
```

**Output**

```python
+--------+-------------------+-----+-------+
|event_id|timestamp          |value|name   |
+--------+-------------------+-----+-------+
|1       |2024-12-10 10:05:00|60   |John   |
|2       |2024-12-10 10:03:00|30   |Alice  |
+--------+-------------------+-----+-------+
```

# Complete Code

```python
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

if __name__ == '__main__':
    # Create Spark session
    spark = SparkSession.builder \
        .appName("Streaming Deduplication and Quality Enforcement") \
        .master("local") \
        .getOrCreate()

    # Define custom schema
    CUSTOM_SCHEMA = StructType([
        StructField("event_id", StringType(), True),    # Unique event identifier
        StructField("name", StringType(), True),       # Name of the entity
        StructField("timestamp", TimestampType(), True),  # Event timestamp
        StructField("value", IntegerType(), True)      # Numeric value for the event
    ])

    # Source: Read streaming data from CSV files
    streaming_df = spark.readStream \
        .format("csv") \
        .schema(CUSTOM_SCHEMA) \
        .load("../resources/dataset/data_quality/input")

    # Deduplicate using time-based windowing
    deduplicated_df = streaming_df.withWatermark("timestamp", "10 minutes") \
        .groupBy(
            window(col("timestamp"), "10 minutes"),  # Group by time window
            col("event_id")                         # Group by event_id
        ) \
        .agg(
            max("timestamp").alias("latest_timestamp"),  # Keep the latest timestamp
            max("value").alias("value"),                # Keep the latest value
            first("name").alias("name")                 # Keep the first non-null name
        ) \
        .select(
            col("event_id"),
            col("latest_timestamp").alias("timestamp"),
            col("value"),
            col("name")
        )

    # Enforce data quality rules
    quality_enforced_df = deduplicated_df \
        .filter((col("value").isNotNull()) & (col("value") > 0)) \
        .fillna({"name": "Unknown"})  # Replace null names

    # Sink: Write the results to the console
    quality_enforced_df.writeStream.start(format="console",
                                          outputMode="complete").awaitTermination()
```