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

spark = SparkSession.builder \
    .appName("DataValidation") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("json") \
    .load("/path/to/data")

# Define validation rules
# Rule 1: No null values in critical columns
# Parameter: isNotNull() — checks that column value is not null
df_not_null = df.filter(
    col("user_id").isNotNull() &
    col("email").isNotNull() &
    col("event_date").isNotNull()
)

# Rule 2: Data type validation
# Parameter: cast() — attempts to cast to target type, null if fails
df_type_valid = df.filter(
    col("amount").cast(DoubleType()).isNotNull() &
    col("event_date").cast(DateType()).isNotNull()
)

# Rule 3: Value range validation
# Parameter: between() — checks if value is within range
df_range_valid = df.filter(
    (col("amount") >= 0) &
    (col("amount") <= 1000000) &
    col("amount").between(0, 1000000)  # Alternative syntax
)

# Rule 4: Format validation using regex
# Parameter: rlike() — checks if string matches regex pattern
df_format_valid = df.filter(
    col("email").rlike(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$") &
    col("phone").rlike(r"^\+?[1-9]\d{1,14}$")  # E.164 format
)

# Combine all validations
df_valid = df.join(df_not_null.select("id"), "id", "inner") \
    .join(df_type_valid.select("id"), "id", "inner") \
    .join(df_range_valid.select("id"), "id", "inner") \
    .join(df_format_valid.select("id"), "id", "inner")

print(f"Original records: {df.count()}")
print(f"Valid records: {df_valid.count()}")
print(f"Validation rate: {df_valid.count() / df.count() * 100:.2f}%")

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

spark = SparkSession.builder \
    .appName("AnomalyDetection") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("delta") \
    .load("/path/to/data")

# Calculate statistics for anomaly detection
# Parameter: mean(), stddev() — aggregate functions for statistics
stats = df.select(
    mean("amount").alias("mean_amount"),
    stddev("amount").alias("stddev_amount"),
    count("amount").alias("count_amount"),
    min("amount").alias("min_amount"),
    max("amount").alias("max_amount")
).collect()[0]

# Apply Z-score anomaly detection
# Parameter: when().otherwise() — conditional column creation
df_with_anomalies = df.withColumn(
    "z_score",
    (col("amount") - lit(stats["mean_amount"])) / lit(stats["stddev_amount"])
).withColumn(
    "is_anomaly",
    when(abs(col("z_score")) > 3.0, True).otherwise(False)  # 3-sigma rule
).withColumn(
    "anomaly_type",
    when(col("z_score") > 3.0, "HIGH_VALUE")
    .when(col("z_score") < -3.0, "LOW_VALUE")
    .otherwise("NORMAL")
)

# Calculate anomaly statistics
anomaly_stats = df_with_anomalies.groupBy("is_anomaly").agg(
    count("*").alias("record_count"),
    avg("amount").alias("avg_amount"),
    stddev("amount").alias("stddev_amount")
)

print("Anomaly Statistics:")
anomaly_stats.show()

# Detect distribution anomalies using quantiles
# Parameter: approx_quantile() — approximate quantile calculation
quantiles = df.select(
    expr("percentile_approx(amount, 0.25)").alias("q1"),
    expr("percentile_approx(amount, 0.50)").alias("median"),
    expr("percentile_approx(amount, 0.75)").alias("q3")
).collect()[0]

iqr = quantiles["q3"] - quantiles["q1"]
lower_bound = quantiles["q1"] - 1.5 * iqr
upper_bound = quantiles["q3"] + 1.5 * iqr

# Mark IQR-based anomalies
df_iqr_anomalies = df.withColumn(
    "iqr_anomaly",
    when(
        (col("amount") < lower_bound) | (col("amount") > upper_bound),
        True
    ).otherwise(False)
)

print(f"IQR bounds: [{lower_bound:.2f}, {upper_bound:.2f}]")
print(f"IQR anomalies: {df_iqr_anomalies.filter(col('iqr_anomaly')).count()}")

from pyspark.sql import SparkSession
from great_expectations.dataset import SparkDFDataset

spark = SparkSession.builder \
    .appName("GreatExpectations") \
    .getOrCreate()

# Read data into Great Expectations
df = spark.read \
    .format("delta") \
    .load("/path/to/data")

# Wrap Spark DataFrame with Great Expectations
ge_df = SparkDFDataset(df)

# Define expectations
# Parameter: expect_column_values_to_not_be_null — null check
ge_df.expect_column_values_to_not_be_null("user_id")

# Parameter: expect_column_values_to_be_unique — uniqueness check
ge_df.expect_column_values_to_be_unique("email")

# Parameter: expect_column_values_to_be_in_set — value domain check
ge_df.expect_column_values_to_be_in_set(
    "status",
    ["active", "inactive", "pending", "suspended"]
)

# Parameter: expect_column_values_to_match_regex — format validation
ge_df.expect_column_values_to_match_regex(
    "email",
    r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
)

# Parameter: expect_column_values_to_be_between — range validation
ge_df.expect_column_values_to_be_between(
    "amount",
    min_value=0,
    max_value=1000000
)

# Parameter: expect_table_row_count_to_be_between — volume check
ge_df.expect_table_row_count_to_be_between(
    min_value=1000,
    max_value=1000000
)

# Run validation
results = ge_df.validate()

print(f"Validation success: {results.success}")
print(f"Successful expectations: {results.statistics['successful_expectations']}")
print(f"Unsuccessful expectations: {results.statistics['unsuccessful_expectations']}")

from pyspark.sql import SparkSession
from com.amazon.deequ import VerificationSuite, VerificationResult
from com.amazon.deequ.checks import Check, CheckLevel
from com.amazon.deequ.constraints import Constraint

spark = SparkSession.builder \
    .appName("DeequValidation") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("delta") \
    .load("/path/to/data")

# Define verification suite with checks
# Parameter: CheckLevel.Error — severity level for check failures
verification_suite = VerificationSuite() \
    .onData(df) \
    .addCheck(
        Check(spark, CheckLevel.Error, "Data Quality Check")
        # Null check
        .isComplete("user_id")
        .isComplete("email")
        .isComplete("event_date")
        
        # Uniqueness check
        .isUnique("email")
        
        # Range check
        .isContainedIn("amount", 0, 1000000)
        
        # Value set check
        .isContainedIn("status", ["active", "inactive", "pending"])
        
        # Compliance check (custom SQL)
        .satisfies(
            "event_date <= current_date()",
            "event_date_not_future",
            CheckLevel.Error,
            "Event date should not be in the future"
        )
        
        # Pattern check
        .matchesPattern(
            "email",
            r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
            "email_format"
        )
    )

# Run verification
result = verification_suite.run()

# Get results
if result.status == VerificationResult.SUCCESS:
    print("All checks passed!")
else:
    print("Some checks failed:")
    for check in result.checkResults:
        if check.status != "SUCCESS":
            print(f"  - {check.check}: {check.status}")

# Get metrics
metrics = result.metrics
print(f"\nMetrics:")
for metric_name, metric_value in metrics.items():
    print(f"  {metric_name}: {metric_value}")