Apache Spark 4.0.1 preview now accessible on Amazon EMR Serverless

Amazon EMR Serverless now helps Apache Spark 4.0.1 in preview, making analytics accessible to extra customers, simplifying information engineering workflows, and strengthening governance capabilities. The discharge introduces ANSI SQL compliance, VARIANT information sorts assist for JSON dealing with, Apache Iceberg v3 desk format assist, and enhanced streaming capabilities. This preview is offered in all areas the place EMR Serverless is offered.

On this put up, we discover key advantages, technical capabilities, and issues for getting began with Spark 4.0.1 on Amazon EMR Serverless—a serverless deployment possibility that simplifies operating open-source large information frameworks, with out requiring managing clusters. With the emr-spark-8.0-preview launch label, you’ll be able to consider new SQL capabilities, Python API enhancements, and streaming enhancements in your present EMR Serverless setting.

Advantages

Spark 4.0.1 helps you remedy information engineering issues with particular enhancements. This part reveals how new capabilities assist with real-world situations.

Make analytics accessible to extra customers

Simplify Extract Remodel Load (ETL) growth with SQL scripting. Knowledge engineers usually change between SQL and Python to construct complicated ETL logic with management movement. SQL scripting in Spark 4.0.1 permits loops, conditionals, and session variables straight in SQL, decreasing context-switching and simplifying pipeline growth. Use pipe syntax (|>) to chain operations for extra readable, maintainable queries.

Enhance information high quality with ANSI SQL mode. Silent kind conversion failures can introduce information high quality points. ANSI SQL mode (now default) enforces commonplace SQL conduct, elevating errors for invalid operations as an alternative of manufacturing surprising outcomes. Necessary: ANSI SQL mode is now enabled by default. Check your queries totally throughout this preview analysis.

Simplify information engineering workflows

Course of JSON information effectively with VARIANT. Groups working with semi-structured information usually see gradual efficiency from repeated JSON parsing. The VARIANT information kind shops JSON in an optimized binary format, eliminating parsing overhead. You may effectively retailer and question JSON information in information lakes with out schema rigidity.

Construct Python information sources with out Scala. Integrating customized information sources beforehand required Scala experience. The Python information Supply API permits you to construct connectors totally in Python, utilizing present Python abilities and libraries with out studying a brand new language.

Debug streaming functions with queryable state. Troubleshooting stateful streaming functions has traditionally required oblique strategies. The brand new state information supply reader reveals streaming state as queryable DataFrames. You may examine state throughout debugging, take a look at state values in unit checks, and diagnose manufacturing incidents.

Strengthen governance capabilities

Set up complete audit trails with Apache Iceberg v3. The Apache Iceberg v3 desk format offers transaction ensures and tracks information adjustments over time, supplying you with the audit trails wanted for regulatory compliance. When mixed with VARIANT information kind assist, you’ll be able to preserve governance controls whereas dealing with semi-structured information effectively in information lakes.

Key capabilities

Spark 4.0.1 Preview on EMR Serverless introduces 4 main functionality areas:

1. SQL enhancements – ANSI mode, pipe syntax, VARIANT kind, SQL scripting, user-defined capabilities (UDFs)
2. Python API advances – customized information sources, UDF profiling
3. Streaming enhancements – stateful processing API v2, queryable state
4. Desk format assist – Amazon S3 Tables, AWS Lake Formation integration

The next sections present technical particulars and code examples for every functionality.

SQL enhancements

Spark 4.0.1 introduces new SQL capabilities together with ANSI mode compliance, SQL UDFs, pipe syntax for readable queries, VARIANT kind for JSON dealing with, and SQL scripting with management movement.

ANSI SQL mode by default

ANSI SQL mode is now enabled by default, imposing commonplace SQL conduct for information integrity. Silent casting of out-of-range values now raises errors relatively than producing surprising outcomes. Current queries could behave in a different way, significantly round null dealing with, string casting, and timestamp operations. Use spark.sql.ansi.enabled=false for those who want legacy conduct throughout migration.

SQL pipe syntax

Now you can chain SQL operations utilizing the |> operator for improved readability. The next instance reveals how one can exchange nested subqueries with a extra maintainable pipeline:

FROM buyer
|> LEFT OUTER JOIN orders ON c_custkey = o_custkey
|> AGGREGATE COUNT(o_orderkey) c_count GROUP BY c_custkey
|> AGGREGATE COUNT(*) AS custdist GROUP BY c_count
|> ORDER BY custdist DESC

This replaces nested subqueries, making complicated transformations simpler to grasp and preserve.

VARIANT information kind

The VARIANT kind handles semi-structured JSON/XML information effectively with out repeated parsing. It makes use of an optimized binary illustration internally whereas sustaining schema-less flexibility. Beforehand, JSON expressions required repeated parsing, degrading efficiency. VARIANT eliminates this overhead. The next snippet reveals parse JSON into the VARIANT kind:

df = spark.sql("SELECT parse_json('{"title":"Alice","age":30}') as information")

Spark 4.0.1 on EMR Serverless helps Apache Iceberg v3, enabling the VARIANT information kind with Iceberg tables. This mixture offers environment friendly storage and querying of semi-structured JSON information in your information lake. Retailer VARIANT columns in Iceberg tables and use Iceberg’s schema evolution and time journey capabilities alongside Spark’s optimized JSON processing. The next instance reveals create an Iceberg desk with a VARIANT column:

CREATE TABLE catalog.db.occasions (
  event_id BIGINT,
  event_data VARIANT,
  timestamp TIMESTAMP
) USING iceberg;

INSERT INTO catalog.db.occasions SELECT 1, parse_json('{"person":"alice","motion":"login"}'), current_timestamp();

SQL scripting with session variables

Handle state and management movement straight in SQL utilizing session variables, and IF/WHILE/FOR statements. The next instance demonstrates a loop that populates a outcomes desk:

BEGIN
  DECLARE counter INT = 10;
  WHILE counter > 0 DO
    INSERT INTO outcomes VALUES (counter);
    SET counter = counter - 1;
  END WHILE;
END

This allows complicated ETL logic totally in SQL with out switching to Python.

SQL user-defined capabilities

Outline customized capabilities straight in SQL. Features may be momentary (session-scoped) or everlasting (catalog-stored). The next instance reveals register and use a easy UDF:

CREATE FUNCTION plusOne(x INT) RETURNS INT RETURN x + 1;
SELECT plusOne(5);

Python API advances

This part covers new Python capabilities together with customized information sources and UDF profiling instruments.

Python information supply API

Now you can construct customized information sources in Python with out Scala data. The next instance reveals create a easy information supply that returns pattern information:

from pyspark.sql.datasource import DataSource, DataSourceReader
from pyspark.sql.sorts import StructType, StructField, StringType, IntegerType

class SampleDataSource(DataSource):
    def schema(self):
        return StructType([
            StructField("name", StringType()),
            StructField("age", IntegerType())
        ])
    
    def reader(self, schema):
        return SampleReader()

class SampleReader(DataSourceReader):
    def learn(self, partition):
        yield ("Alice", 30)
        yield ("Bob", 25)

# Register and use
spark.dataSource.register(SampleDataSource)
spark.learn.format("SampleDataSource").load().present()

Unified UDF profiling

Profile Python and Pandas UDFs for efficiency and reminiscence insights. The next code permits efficiency profiling:spark.conf.set("spark.sql.pyspark.udf.profiler", "perf") # or "reminiscence"

Structured streaming enhancements

This part covers enhancements to stateful stream processing, together with queryable state and enhanced state administration APIs.

Arbitrary stateful processing API v2

The transformWithState operator offers strong state administration with timer and TTL assist for automated cleanup, schema evolution capabilities, and preliminary state assist for pre-populating state from batch DataFrames.

State information supply reader

Question streaming state as a DataFrame for debugging and monitoring. Beforehand, state information was inner to streaming queries. Now you’ll be able to confirm state values in unit checks, diagnose manufacturing incidents, detect state corruption, and optimize efficiency. Observe: This function is experimental. Supply choices and conduct could change in future releases.

State retailer enhancements

Upgraded changelog checkpointing for RocksDB removes efficiency bottlenecks. Enhanced checkpoint coordination and improved sorted string desk (SST) file reuse administration optimizes streaming operations.

Desk format assist

This part covers assist for AWS S3 Tables and full desk entry (FTA) with AWS Lake Formation.

AWS S3 Tables

Use Spark 4.0.1 with AWS S3 Tables, a storage resolution that gives managed Apache Iceberg tables with automated optimization and upkeep. S3 Tables simplify information lake operations by dealing with compaction, snapshot administration, and metadata cleanup robotically.

Full desk entry with Lake Formation

FTA is supported for Apache Iceberg, Delta Lake, and Apache Hive tables when utilizing AWS Lake Formation, a managed service that simplifies information entry management. FTA offers coarse-grained entry management on the desk degree. Observe that fine-grained entry management (FGAC) with column-level or row-level permissions is just not accessible on this preview.

Getting began

Comply with these steps to create an EMR Serverless software, run pattern code to check new options, and supply suggestions on the preview.

Stipulations

Earlier than you start, verify you’ve gotten the next:

Observe: EMR Studio Notebooks and SageMaker Unified Studio should not supported throughout this preview. Use the AWS CLI or AWS SDK to submit jobs.

Step 1: Create your EMR Serverless software

Create or replace your software with the emr-spark-8.0-preview launch label. The next command creates a brand new software:

aws emr-serverless create-application --type spark 
  --release-label emr-spark-8.0-preview 
  --region us-east-1 --name spark4-test

Step 2: Check pattern code

Run this PySpark job to confirm setup and take a look at Spark 4.0.1 options:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Spark 4.0.1 Check").getOrCreate()
print(f"Spark Model: {spark.model}")

# Create pattern information
information = [("Alice", 34, "Engineering"), ("Bob", 45, "Sales"),
        ("Charlie", 28, "Engineering"), ("Diana", 52, "Marketing")]
df = spark.createDataFrame(information, ["name", "age", "department"])
df.createOrReplaceTempView("staff")

# Check SQL PIPE syntax
attempt:
    consequence = spark.sql("""
        FROM staff
        |> WHERE age > 30
        |> SELECT title, age, division
        |> ORDER BY age DESC
    """)
    consequence.present()
    print("✓ SQL pipe syntax take a look at handed")
besides Exception as e:
    print(f"✗ SQL pipe syntax take a look at failed: {e}")

# Check VARIANT information kind
attempt:
    json_data = spark.sql("""
        SELECT parse_json('{"title":"Alice","abilities":["Python","Spark","SQL"]}') as information
    """)
    json_data.present(truncate=False)
    print("✓ VARIANT information kind take a look at handed")
besides Exception as e:
    print(f"✗ VARIANT information kind take a look at failed: {e}")

Submit the job with the next command:

aws emr-serverless start-job-run 
    --application-id  
    --execution-role-arn  
    --job-driver '{
        "sparkSubmit": {
            "entryPoint": "s3:///spark_4_test.py",
            "sparkSubmitParameters": "--conf spark.executor.cores=4 --conf spark.executor.reminiscence=16g"
        }
    }'

Step 3: Check your workloads

Overview the Spark SQL Migration Information and PySpark Migration Information, then take a look at manufacturing workloads in non-production environments. Concentrate on queries affected by ANSI SQL mode and benchmark efficiency.

Step 4: Clear up sources

After testing, delete all sources created throughout this analysis to keep away from ongoing costs:

# Delete the EMR Serverless software
aws emr-serverless delete-application 
    --application-id spark4-test 
    --region us-east-1
# Take away the take a look at script from S3
aws s3 rm s3:///spark_4_test.py

Migration issues

Earlier than evaluating Spark 4.0.1, overview the up to date runtime necessities and behavioral adjustments which will have an effect on your present code.

Runtime necessities

• Scala: Model 2.13.16 required (2.12 assist dropped)
• Java: JDK 17 or larger required (JDK 8 and 11 assist eliminated)
• Python: Model 3.9+ required, continued assist for 3.11 and newly added 3.12 (3.8 assist eliminated)
• Pandas: Minimal model 2.0.0 (beforehand 1.0.5)
• SparkR: Deprecated; migrate to PySpark

Behavioral adjustments

With ANSI SQL mode enforcement, you might even see totally different conduct in:

• Null dealing with: Stricter null propagation in expressions
• String casting: Invalid casts now elevate errors as an alternative of returning null
• Map key operations: Duplicate keys now elevate errors
• Timestamp conversions: Overflow returns null as an alternative of wrapped values
• CREATE TABLE statements: Now respect the spark.sql.sources.defaultconfiguration as an alternative of defaulting to Hive format when USING or STORED AS clauses are omitted

You may management many of those behaviors through legacy configuration flags. Seek the advice of the official migration guides for particulars refer: Spark SQL Migration Information: 3.5 to 4.0 and PySpark Migration Information: 3.5 to 4.0.

Preview limitations

The next capabilities should not accessible on this preview:

• High quality-grained entry management: High quality-grained entry management (FGAC) with row-level or column-level filtering is just not supported on this preview. Jobs with spark.emr-serverless.lakeformation.enabled=true will fail.
• Spark Join: Not supported on this preview. Use commonplace Spark job submission with the StartJobRun API.
• Open Desk Format limitations: Hudi is just not supported on this preview. Delta 4.0.0 doesn’t assist Flink connectors (deprecated in Delta 4.0.0). Delta Common Format is just not supported on this preview.
• Connectors: spark-sql-kinesis, emr-dynamodb, and spark-redshift are unavailable.
• Interactive functions: Livy and JupyterEnterpriseGateway should not included. Additionally, SageMaker Unified Studio and EMR Studio should not supported.
• EMR options: Serverless Storage and Materialized Views should not supported.

This preview permits you to consider Spark 4.0.1’s core capabilities on EMR Serverless, together with SQL enhancements, Python API enhancements, and streaming state administration. Check your migration path, assess efficiency enhancements, and supply suggestions to form the final availability launch.

Conclusion

This put up confirmed you get began with the Apache Spark 4.0.1 preview launch on Amazon EMR Serverless. You explored how the VARIANT information kind works with Iceberg v3 to course of JSON information effectively, how SQL scripting and pipe syntax remove context-switching for ETL growth, and the way queryable streaming state simplifies debugging stateful functions. You additionally realized in regards to the preview limitations, runtime necessities, and behavioral adjustments to think about throughout analysis.

Check the Spark 4.0.1 preview on EMR Serverless and supply suggestions by AWS Assist to assist form the final availability launch.

To be taught extra about Apache Spark 4.0.1 options, see the Spark 4.0.1 Launch Notes. For EMR Serverless documentation, see the EMR Launch Information.

Sources

Apache Spark Documentation

Amazon EMR Sources

In regards to the authors