Typical clue in a failing PySpark job's log:

java.lang.OutOfMemoryError: Java heap space
    at java.util.Arrays.copyOf(Arrays.java:...)
    at org.apache.spark.sql.execution...

What it tells you:
- The JVM executor ran out of heap memory
- Look at partition sizes, skewed keys, collect() calls
- Consider spark.executor.memory or more partitions

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class CityTotals {
    record Order(String city, long amountPaise) {}

    public static void main(String[] args) {
        List<Order> orders = List.of(
            new Order("Noida", 120000), new Order("Delhi", 80000), new Order("Noida", 50000));
        Map<String, Long> totals = orders.stream()
            .collect(Collectors.groupingBy(Order::city, Collectors.summingLong(Order::amountPaise)));
        System.out.println(totals);   // e.g. {Delhi=80000, Noida=170000}
    }
}

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class OrderProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");
        props.put("enable.idempotence", "true");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            String value = "{\"order_id\":\"A1\",\"amount_paise\":120000}";
            producer.send(new ProducerRecord<>("orders", "customer-42", value));
        }
    }
}

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class OrderConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-loader");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));
            while (true) {
                for (ConsumerRecord<String, String> r : consumer.poll(Duration.ofSeconds(1))) {
                    System.out.printf("partition=%d offset=%d value=%s%n", r.partition(), r.offset(), r.value());
                }
                consumer.commitSync();   // commit after processing the batch
            }
        }
    }
}

from collections import defaultdict

orders = [("Noida", 120000), ("Delhi", 80000), ("Noida", 50000)]
totals = defaultdict(int)
for city, paise in orders:
    totals[city] += paise
print(dict(totals))   # {'Noida': 170000, 'Delhi': 80000}

import org.apache.spark.sql.*;
import static org.apache.spark.sql.functions.*;

public class DailySales {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("daily-sales").getOrCreate();
        Dataset<Row> orders = spark.read().option("header", "true").option("inferSchema", "true")
            .csv("data/orders.csv");
        orders.groupBy(col("city"))
              .agg(sum("amount_paise").alias("total_paise"))
              .orderBy(desc("total_paise"))
              .show();
        spark.stop();
    }
}

Weeks 1-2: Java syntax, types, classes, records, collections
Week 3:    Streams API, exceptions, Maven or Gradle builds
Week 4:    JVM basics — heap, garbage collection, reading stack traces
Week 5:    Kafka producer and consumer in Java
Week 6:    Kafka Streams or Flink: a simple aggregation
Week 7:    Spark Dataset API job in Java
Week 8:    A documented streaming mini-project

<!-- pom.xml dependencies (check for current versions) -->
<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.7.0</version>
  </dependency>
  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>2.0.13</version>
  </dependency>
</dependencies>

Streaming mini-project
☐ Java producer sends order events with customer keys
☐ Consumer group with two instances shares partitions
☐ Commits offsets after processing
☐ Idempotent writes to a database table (upsert by order_id)
☐ README: delivery guarantee explained, how to run