sentiment_LR.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when
from pyspark.ml import Pipeline
from pyspark.ml.feature import (
    RegexTokenizer, StopWordsRemover,
    HashingTF, IDF, NGram, VectorAssembler
)
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark import StorageLevel
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.functions import col


# ============================================================
# CONFIGURATION
# ============================================================
INPUT_PATH = "hdfs://localhost:9000/tweetanalysis/processed/cleaned_tweets"
MODEL_PATH = "hdfs://localhost:9000/tweetanalysis/models/sentiment_lr"


# ============================================================
# SPARK SESSION
# ============================================================
def create_spark_session():
    spark = (
        SparkSession.builder
        .appName("Sentiment_LR_Optimized")
        .master("spark://localhost:7077")
        .config("spark.sql.shuffle.partitions", "8")
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("ERROR")
    return spark


# ============================================================
# DATA LOADING & LABELING
# ============================================================
def load_and_label_data(spark):
    df = spark.read.parquet(INPUT_PATH)

    df = df.withColumn(
        "label",
        when(col("sentiment") == 4, 1.0).otherwise(0.0)
    )

    df.persist(StorageLevel.MEMORY_AND_DISK)

    return df


# ============================================================
# FEATURE PIPELINE
# ============================================================
def build_feature_pipeline():
    tokenizer = RegexTokenizer(
        inputCol="cleaned_text",
        outputCol="words",
        pattern="\\W+"
    )

    stopwords = StopWordsRemover(
        inputCol="words",
        outputCol="filtered_words"
    )

    tf = HashingTF(
        inputCol="filtered_words",
        outputCol="tf_features",
        numFeatures=8000
    )

    idf = IDF(
        inputCol="tf_features",
        outputCol="tfidf_features",
        minDocFreq=5
    )

    bigram = NGram(
        n=2,
        inputCol="filtered_words",
        outputCol="bigrams"
    )

    bigram_tf = HashingTF(
        inputCol="bigrams",
        outputCol="bigram_features",
        numFeatures=8000
    )

    assembler = VectorAssembler(
        inputCols=["tfidf_features", "bigram_features"],
        outputCol="features"
    )

    return [
        tokenizer,
        stopwords,
        tf,
        idf,
        bigram,
        bigram_tf,
        assembler
    ]


# ============================================================
# MODEL
# ============================================================
def build_model():
    return LogisticRegression(
        featuresCol="features",
        labelCol="label",
        family="binomial",
        maxIter=30,
        regParam=0.001
    )


# ============================================================
# TRAINING
# ============================================================
def train_model(df):
    train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)

    stages = build_feature_pipeline()
    stages.append(build_model())

    pipeline = Pipeline(stages=stages)

    model = pipeline.fit(train_df)
    return model, test_df



# ============================================================
# EVALUATION & RESULT PRINTING
# ============================================================
def evaluate_model(model, test_df):
    predictions = model.transform(test_df)

    print("\n" + "=" * 60)
    print("SENTIMENT MODEL EVALUATION RESULTS")
    print("=" * 60)

    metrics = {
        "Accuracy": "accuracy",
        "Precision": "weightedPrecision",
        "Recall": "weightedRecall",
        "F1-Score": "f1"
    }

    for name, metric in metrics.items():
        evaluator = MulticlassClassificationEvaluator(
            labelCol="label",
            predictionCol="prediction",
            metricName=metric
        )
        score = evaluator.evaluate(predictions)
        print(f"{name:<15}: {score:.4f}")

    print("=" * 60)

    # --------------------------------------------------------
    # Confusion Matrix
    # --------------------------------------------------------
    print("\nCONFUSION MATRIX (label vs prediction)")
    predictions.groupBy("label", "prediction").count().orderBy("label", "prediction").show()

    # --------------------------------------------------------
    # Prediction Distribution
    # --------------------------------------------------------
    print("PREDICTION DISTRIBUTION")
    predictions.groupBy("prediction").count().show()

    # --------------------------------------------------------
    # Sample Predictions
    # --------------------------------------------------------
    print("SAMPLE PREDICTIONS")
    predictions.select(
        col("cleaned_text"),
        col("label"),
        col("prediction")
    ).show(10, truncate=60)

    return predictions


# ============================================================
# SAVE MODEL
# ============================================================
def save_model(model):
    model.write().overwrite().save(MODEL_PATH)
    print(f"Model saved to: {MODEL_PATH}")


# ============================================================
# MAIN
# ============================================================
def main():
    spark = create_spark_session()

    try:
        df = load_and_label_data(spark)
        model, test_df = train_model(df)
        evaluate_model(model, test_df)
        save_model(model)

    finally:
        spark.stop()


if __name__ == "__main__":
    main()