SparkAid

Common errors

• AttributeError: module 'sparkaid' has no attribute 'flatten'

  cause The `flatten` function (or other specific utilities) was not correctly imported from the `sparkaid` library.
  fix Ensure you are importing the specific utility directly, e.g., `from sparkaid import flatten`.

• org.apache.spark.SparkException: Job aborted due to stage failure: Task XXX lost. ... java.lang.OutOfMemoryError

  cause This is a general Spark error, often caused by trying to process too much data in memory, especially when flattening very wide or deep nested schemas, or performing operations that require significant data shuffling or aggregation.
  fix Increase executor/driver memory, repartition data to avoid skewed partitions, or review your data processing logic to minimize memory-intensive operations. Break down complex flattening tasks if necessary.

• Py4JJavaError: An error occurred while calling oX.schema. ... Spark encountered a serialization error

  cause Spark requires objects to be serializable to be sent across the network to executors. This error often occurs with user-defined functions (UDFs) or closures that capture non-serializable objects, or when data structures manipulated by `sparkaid` become too complex for default serialization.
  fix Ensure all objects referenced within UDFs are serializable. For complex schemas, consider simplifying the data before broad transformations or investigate custom serialization if standard Spark serialization fails.

Warnings

• breaking In SparkAid v1.0.0, the `flatten()` function's default behavior changed. It now stops unpacking nested data at `ArrayType` fields. To flatten elements within arrays, you must explicitly provide the `arrays_to_unpack=["*"]` parameter or specify the array columns.
  Fix: For previous behavior where all nested array elements were flattened, update your `flatten()` calls to include `arrays_to_unpack=["*"]` or `arrays_to_unpack=['your_array_column']`.
• gotcha When using SparkAid (or any PySpark code), be aware of common Spark performance anti-patterns, such as calling `collect()` or `toPandas()` on large DataFrames, misconfiguring shuffle operations, or improper caching. These can lead to `OutOfMemoryError` or slow job execution.
  Fix: Avoid `collect()` for large datasets; use actions like `take()` or write directly to storage. Tune `spark.sql.shuffle.partitions`. Use `.cache()` or `.persist()` judiciously and `unpersist()` when no longer needed. Analyze and address data skew.
• breaking If upgrading your underlying Apache Spark version, especially to Spark 4.0+, be aware of significant breaking changes in Spark itself, such as default ANSI SQL mode, Java 17 requirement, and Hadoop 3.3.6+ requirement. These can impact any PySpark application, including those using SparkAid.
  Fix: Review Spark migration guides, especially for SQL and PySpark. Ensure your Java and Hadoop environments meet the new requirements. Test applications thoroughly with `spark.sql.ansi.enabled=false` if strict ANSI mode causes issues.

Install

• pip install sparkaid Install latest version

Imports

• flatten

  from sparkaid import flatten

Quickstart

Demonstrates how to initialize a SparkSession, create a DataFrame with nested structures, and use `sparkaid.flatten` to unnest the schema. Note the use of `arrays_to_unpack` for array types.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, IntegerType
from sparkaid import flatten

# Initialize Spark Session
spark = SparkSession.builder \
    .appName("SparkAidQuickstart") \
    .master("local[*]") \
    .getOrCreate()

# Create a sample DataFrame with nested structure
data = [
    ("Alice", {"city": "New York", "zip": 10001}, ["apple", "banana"]),
    ("Bob", {"city": "Los Angeles", "zip": 90001}, ["orange"]),
    ("Charlie", None, ["grape", "kiwi", "mango"])
]

schema = StructType([
    StructField("name", StringType(), True),
    StructField("address", StructType([
        StructField("city", StringType(), True),
        StructField("zip", IntegerType(), True)
    ]), True),
    StructField("fruits", ArrayType(StringType()), True)
])

df = spark.createDataFrame(data, schema)
print("Original Schema:")
df.printSchema()

print("\nFlattening DataFrame:")
# Flatten the DataFrame. By default, it flattens StructTypes.
# For array flattening, 'arrays_to_unpack=["*"]' is needed as per v1.0.0 breaking change.
flattened_df = flatten(df, nested_struct_separator="__", arrays_to_unpack=["fruits"])

print("Flattened Schema:")
flattened_df.printSchema()
print("Flattened Data:")
flattened_df.show()

spark.stop()