Supported APIs and compatibility notes

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/"))

Query Engine

Get started

Connect data

Run clusters

Query

Connect your tools

Secure and govern

Troubleshoot

Reference

Concepts

Supported APIs and compatibility notes

DataFrame operations

Transformations

Actions

Temporary views

Read formats

Write modes

SQL functions

Column references and literals

String functions

Math functions

Aggregate functions

Date and time functions

Conditional functions

JSON, array, and other functions

Code samples

Basic analytics pipeline

Window functions: rankings and running totals

Multi-table join

CASE WHEN logic

Spatial analysis: points in polygons

SQL via temp views, and export to Pandas

Writing results

See also

​DataFrame operations

​Transformations

​Actions

​Temporary views

​Read formats

​Write modes

​SQL functions

​Column references and literals

​String functions

​Math functions

​Aggregate functions

​Date and time functions

​Conditional functions

​JSON, array, and other functions

​Code samples

​Basic analytics pipeline

​Window functions: rankings and running totals

​Multi-table join

​CASE WHEN logic

​Spatial analysis: points in polygons

​SQL via temp views, and export to Pandas

​Writing results

​See also

DataFrame operations

Transformations

Actions

Temporary views

Read formats

Write modes

SQL functions

Column references and literals

String functions

Math functions

Aggregate functions

Date and time functions

Conditional functions

JSON, array, and other functions

Code samples

Basic analytics pipeline

Window functions: rankings and running totals

Multi-table join

CASE WHEN logic

Spatial analysis: points in polygons

SQL via temp views, and export to Pandas

Writing results

See also