Supported APIs and compatibility notes

The PySpark DataFrame operations, SQL functions, and read/write formats supported by the e6data compatibility layer, with end-to-end code samples.

This page lists the DataFrame operations, SQL functions, and file formats supported by e6-spark-compat, followed by end-to-end code samples. For setup, see PySpark compatibility.

DataFrame operations

Transformations are lazy - they build a query plan without executing. Execution is triggered only when an action is called.

Transformations

Method	Description
select(*cols)	Select columns / expressions
selectExpr(*exprs)	Select using SQL expressions
filter(condition) / where()	Filter rows
join(other, on, how)	Join DataFrames
crossJoin(other)	Cross join
groupBy(*cols)	Group for aggregation
orderBy(*cols) / sort()	Sort rows
limit(n)	Limit to first n rows
distinct()	Remove duplicate rows
union(other)	Union (all)
unionByName(other)	Union matching by name
intersect(other)	Set intersection
exceptAll(other)	Set difference
withColumn(name, col)	Add / replace a column
withColumnRenamed(old, new)	Rename a column
drop(*cols)	Drop columns
cache() / persist()	Cache hint (pass-through)
coalesce(n)	Repartition hint

Supported join types: inner, left, right, full, cross, left_semi, left_anti.

# Filter, select, group, and aggregate
df.filter(col("age") > 21) \
  .select("name", "city", "salary") \
  .groupBy("city") \
  .agg(count("*").alias("total"), avg("salary").alias("avg_salary"))

# Pivot
df.groupBy("year").pivot("quarter").sum("revenue")

Actions

Actions trigger query execution on e6data and return results.

Method	Description
collect()	Return all rows as a list of Rows
count()	Return the total row count
show(n)	Print the first n rows (default 20)
first()	Return the first row
head(n)	Return the first n rows
take(n)	Return the first n rows as a list
toPandas()	Convert results to a Pandas DataFrame
explain()	Print the generated SQL query
describe(*cols)	Compute summary statistics

Temporary views

df.createOrReplaceTempView("employees")
result = spark.sql("SELECT department, COUNT(*) FROM employees GROUP BY department")

df.createGlobalTempView("global_employees")   # accessible across sessions

Read formats

Format	Method
Parquet	spark.read.parquet(path)
ORC	spark.read.orc(path)
CSV	spark.read.csv(path)
JSON	spark.read.json(path)
GeoParquet	spark.read.format("geoparquet").load(path)
GeoJSON	spark.read.format("geojson").load(path)
Delta	spark.read.format("delta").load(path)
Text	spark.read.text(path)

df = spark.read.format("csv").option("header", True).option("inferSchema", True).load("s3://bucket/data.csv")

Write modes

Use df.write to save results - .parquet(), .csv(), .partitionBy(...), .insertInto("table"), or .saveAsTable("table").

Mode	Behavior
error	Throw an error if data already exists (default)
append	Append to existing data
overwrite	Overwrite existing data
ignore	Silently skip if data already exists

SQL functions

All functions are imported from e6_spark_compat.sql.functions.

Column references and literals

Function	Description
col(name)	Reference a column
lit(value)	Create a literal value
expr(sql)	Raw SQL expression

String functions

upper, lower, trim, ltrim, rtrim, length, substring(col, pos, len), concat(*cols), concat_ws(sep, *cols), split(col, pattern), regexp_extract(col, pattern, idx), regexp_replace(col, pattern, rep), translate, lpad, rpad, repeat, reverse, format_string, format_number.

Math functions

abs, round(col, scale), floor, ceil, sqrt, pow(col, exp), log, log10, exp, sin, cos, tan, asin, acos, atan, atan2, greatest(*cols), least(*cols).

Aggregate functions

count, countDistinct(*cols), sum, avg, min, max, first, last, collect_list, collect_set, stddev, variance, approx_percentile(col, pct).

Date and time functions

Function	Description
current_date() / current_timestamp()	Current date / timestamp
year, month, day / dayofmonth, hour, minute, second	Extract date parts
date_add(col, days) / date_sub(col, days)	Add / subtract days
datediff(end, start)	Difference in days
to_date(col, fmt) / to_timestamp(col, fmt)	Convert to date / timestamp
from_unixtime(col, fmt) / unix_timestamp(col, fmt)	Unix-timestamp conversions

Conditional functions

when(condition, value) (chainable with .when(...) / .otherwise(...)), coalesce(*cols), isnull, isnan, if_(condition, true, false).

JSON, array, and other functions

to_json, from_json(col, schema), get_json_object(col, path); explode(col); broadcast(df), cast(col, type).

Window functions are supported via the full Window specification API (partitionBy, orderBy, rowsBetween, rangeBetween) with row_number, rank, lag, lead, and aggregates over a window.

Code samples

Basic analytics pipeline

from e6_spark_compat.sql.functions import col, count, sum, avg

orders = spark.read.parquet("s3://data-lake/orders/")

summary = (orders
    .filter(col("status") == "completed")
    .groupBy("region")
    .agg(
        count("*").alias("order_count"),
        sum("amount").alias("total_revenue"),
        avg("amount").alias("avg_order_value"),
    )
    .orderBy(col("total_revenue").desc()))

summary.show()

Window functions: rankings and running totals

from e6_spark_compat.sql.functions import col, row_number, sum, lag
from e6_spark_compat.sql.window import Window

rank_window = Window.partitionBy("department").orderBy(col("salary").desc())
running_window = (Window.partitionBy("department")
    .orderBy("hire_date")
    .rowsBetween(Window.UNBOUNDED_PRECEDING, Window.CURRENT_ROW))

result = employees.select(
    "name", "department", "salary",
    row_number().over(rank_window).alias("salary_rank"),
    lag("salary", 1).over(rank_window).alias("prev_salary"),
    sum("salary").over(running_window).alias("cumulative_salary"),
)

Multi-table join

result = (orders
    .join(customers, orders["customer_id"] == customers["id"], "inner")
    .join(products, orders["product_id"] == products["id"], "inner")
    .groupBy("customers.name", "products.category")
    .agg(count("*").alias("order_count"), sum("orders.amount").alias("total_spent"))
    .orderBy(col("total_spent").desc()))

CASE WHEN logic

from e6_spark_compat.sql.functions import col, when

categorized = transactions.select(
    "*",
    when(col("amount") >= 1000, "high")
        .when(col("amount") >= 100, "medium")
        .otherwise("low")
        .alias("value_tier"),
)

Spatial analysis: points in polygons

from e6_spark_compat.sedona import SedonaRegistrator
from e6_spark_compat.sql.functions import expr, count

SedonaRegistrator.registerAll(spark)

store_regions = stores.join(
    regions,
    expr("ST_Contains(regions.boundary, ST_Point(stores.longitude, stores.latitude))"),
    "inner",
)

SQL via temp views, and export to Pandas

orders.createOrReplaceTempView("orders")
customers.createOrReplaceTempView("customers")

result = spark.sql("""
    SELECT c.name, COUNT(o.id) AS order_count, SUM(o.amount) AS total
    FROM orders o JOIN customers c ON o.customer_id = c.id
    WHERE o.status = 'completed'
    GROUP BY c.name ORDER BY total DESC LIMIT 10
""")

pdf = result.toPandas()   # convert to Pandas for local analysis or plotting

Writing results

daily_totals = orders.groupBy("order_date", "region").agg(sum("amount").alias("daily_total"))

(daily_totals.write
    .mode("overwrite")
    .partitionBy("region")
    .parquet("s3://data-lake/daily-totals/"))

See also

• PySpark compatibility - install and getting started.
• Limitations