One query, six ways

totals = (spark.read.parquet("s3://lake/orders").rdd       # RDD of Row objects
          .filter(lambda r: r.country == "AE")
          .map(lambda r: (r.segment, r.amount))
          .reduceByKey(lambda a, b: a + b))                 # map-side combine, then shuffle
totals.collect()

• The scan reads all 42 columns of every row: Spark can't see that the lambdas only use three.
• The filter isn't pushed down, so no Parquet row groups are skipped.
• Every row is pickled, sent to a Python worker, filtered and mapped there, then sent back.
• reduceByKey still combines on the map side before the shuffle, so at least that part is efficient.

from pyspark.sql import functions as F

out = (spark.read.parquet("s3://lake/orders")
       .filter(F.col("country") == "AE")
       .groupBy("segment")
       .agg(F.sum("amount").alias("total")))
out.show()

• Reads only 3 columns (country, segment, amount).
• The filter is pushed into the Parquet reader, which skips row groups whose statistics rule out "AE".
• A partial_sum runs before the shuffle, and the whole stage is compiled to one generated function.
• No Python involved at run time: Python only described the plan.

spark.read.parquet("s3://lake/orders").createOrReplaceTempView("orders")

out = spark.sql("""
    SELECT segment, SUM(amount) AS total
    FROM orders
    WHERE country = 'AE'
    GROUP BY segment
""")

• Parses into the same logical plan as the DataFrame version, so out.explain() prints the same physical plan.
• Pick SQL or the DataFrame API for readability and team habits, not performance.

import spark.implicits._
case class Order(cust_id: Long, segment: String, country: String, amount: Double)

val ds = spark.read.parquet("s3://lake/orders").as[Order]

// typed style: compile-time checked, but lambdas are opaque to Catalyst
val typed = ds.filter(_.country == "AE")
              .groupByKey(_.segment)
              .mapValues(_.amount)
              .reduceGroups(_ + _)

// column style on the same Dataset: fully optimised
val fast = ds.filter($"country" === "AE")
             .groupBy("segment")
             .agg(sum("amount").as("total"))

• The typed version catches mistakes like _.countrry at compile time, but each row is deserialised into an Order object for the lambda, and the filter can't be pushed down.
• The column version gets the full Catalyst treatment. Many teams use typed Datasets at pipeline boundaries and column expressions inside.

import pandas as pd

pdf = pd.read_parquet("orders.parquet", columns=["segment", "country", "amount"])
out = pdf[pdf["country"] == "AE"].groupby("segment")["amount"].sum()

• Each line runs immediately. No cluster, no plan, no shuffle: fastest of all when the data fits in one machine's memory.
• Selecting columns on read matters here: pandas has no optimiser to do it for you.
• Good fit for small reference data, results after Spark has aggregated, or feeding plotting and ML libraries.

import pyspark.pandas as ps

psdf = ps.read_parquet("s3://lake/orders")
out = psdf[psdf["country"] == "AE"].groupby("segment")["amount"].sum()
out.head()                                    # runs a Spark job

• The same syntax as pandas, but each call is translated into a Spark DataFrame plan and runs distributed.
• It keeps a pandas-style index. Creating a sequential default index costs an extra pass over the data; pass index_col, or set compute.default_index_type to distributed when you don't need consecutive values.
• Useful for moving existing pandas code to big data with few changes. For new pipelines, the DataFrame API is clearer about what Spark does.