Apache
Spark

// Scala – Compute average salary by department
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder()
  .appName("SalaryAnalysis")
  .getOrCreate()

import spark.implicits._

val employees = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv("hdfs:///data/employees.csv")

val avgSalary = employees
  .groupBy("department")
  .agg(avg("salary").alias("avg_salary"),
       count("*").alias("headcount"))
  .orderBy(desc("avg_salary"))

avgSalary.show()

-- Spark SQL – Window functions & CTEs
CREATE OR REPLACE TEMP VIEW sales AS
SELECT * FROM parquet.`s3://bucket/sales/`;

WITH ranked AS (
  SELECT
    region,
    product,
    revenue,
    RANK() OVER (
      PARTITION BY region
      ORDER BY revenue DESC
    ) AS rnk
  FROM sales
  WHERE year = 2024
)
SELECT * FROM ranked
WHERE rnk <= 3
ORDER BY region, rnk;

# Structured Streaming – Kafka → Aggregation → Console
from pyspark.sql.functions import from_json, col, window
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
    .add("user", StringType()) \
    .add("event", StringType()) \
    .add("amount", DoubleType()) \
    .add("ts", StringType())

raw = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "transactions") \
    .load()

events = raw.select(from_json(col("value").cast("string"), schema).alias("d")).select("d.*")

agg = events \
    .withWatermark("ts", "10 minutes") \
    .groupBy(window(col("ts"), "5 minutes"), "event") \
    .sum("amount")

query = agg.writeStream.format("console").start()
query.awaitTermination()

# MLlib – Classification Pipeline
from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, HashingTF, IDF
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Training data: (label, text)
data = spark.createDataFrame([
    (1.0, "spark is fast and scalable"),
    (0.0, "hadoop mapreduce is slow"),
    (1.0, "in-memory computation wins"),
], ["label", "text"])

tokenizer = Tokenizer(inputCol="text", outputCol="words")
tf        = HashingTF(inputCol="words", outputCol="rawFeatures")
idf       = IDF(inputCol="rawFeatures", outputCol="features")
lr        = LogisticRegression(maxIter=10)

pipeline = Pipeline(stages=[tokenizer, tf, idf, lr])
model    = pipeline.fit(data)

evaluator = BinaryClassificationEvaluator()
print(f"AUC: {evaluator.evaluate(model.transform(data)):.3f}")