Accelerate batch workloads and interactive sessions with Native Query Execution

This document describes when and how to enable Native Query Execution to accelerate Serverless for Apache Spark batch workloads and interactive sessions.

Native Query Execution requirements

Serverless for Apache Spark Native Query Execution is available only with batch workloads and interactive sessions using 1.2.26+, 2.2.26+, or a later Spark runtime version running in the Serverless for Apache Spark premium pricing tier. Premium tier pricing is charged at a higher cost than standard tier pricing, but there is no additional charge for Native Query Execution. For pricing information, see Serverless for Apache Spark pricing.

Native Query Execution properties

This section lists required and optional Spark resource allocation properties you can use to enable and customize Native Query Execution for your batch workload or interactive session.

Required property settings

• spark.dataproc.runtimeEngine=native: The workload runtime engine must be set to native to override the default spark runtime engine.

• spark.dataproc.spark.driver.compute.tier=premium and spark.dataproc.executor.compute.tier=premium: These pricing tier properties must be set to the premium pricing tier.

Optional resource allocation properties

• spark.dataproc.driver.disk.tier, spark.dataproc.driver.disk.size, spark.dataproc.executor.disk.tier and spark.dataproc.executor.disk.size: Use these properties to set and configure the premium disk tier and size for the Spark driver and executor processes.

  Premium disk tiers use columnar instead of row-based shuffle to provide better performance. For better shuffle I/O throughput, use the driver and executor premium disk tiers with a sufficiently large disk size to accommodate shuffle files.

• spark.driver.memory, spark.driver.memoryOverhead, spark.executor.memory, spark.executor.memoryOverhead, and spark.memory.offHeap.size: Use these properties to tune memory provided to Spark driver and executor processes.

  You can configure memory in either of the following ways:

  • Option 1: Configure off-heap memory only (spark.memory.offHeap.size) with a specified value. Native Query Execution will use the specified value as off-heap memory, and allocate an additional 1/7th of the off-heap memory value as on-heap memory (spark.executor.memory).

  • Option 2: Configure both on-heap memory (spark.executor.memory) and off-heap memory (spark.memory.offHeap.size). The amount you allocate to off-heap memory must be greater than the amount you allocate to on-heap memory.

  If you don't configure both off-heap memory (spark.memory.offHeap.size) and on-heap memory (spark.executor.memory), the Native Query Execution engine divides a default 4g amount of memory in a 6:1 ratio between off-heap and on-heap memory.

  Recommendation: Allocate off-heap to on-heap memory in a 6:1 ratio.

  Examples:

  Memory settings without Native Query Execution	Recommended memory settings with Native Query Execution
  spark.executor.memory	spark.memory.offHeap.size	spark.executor.memory
  7g	6g	1g
  14g	12g	2g
  28g	24g	4g
  56g	48g	8g

Run the qualification tool

To identify batch workloads that can achieve faster runtimes with Native Query Execution (NQE), you can use the qualification tool. The tool analyzes Spark event logs to estimate potential runtime savings and identify any operations that are not supported by the NQE engine.

Google Cloud provides two methods for running the qualification analysis: qualification job and qualification script. The recommended approach for most users is the qualification job, which automates the discovery and analysis of batch workloads. The alternative qualification script is available for the specific use case of analyzing a known event log file. Choose the method that best fits your use case:

• Qualification Job (Recommended): This is the primary and recommended method. It is a PySpark job that automatically discovers and analyzes recent batch workloads across one or more Google Cloud projects and regions. Use this method when you want to perform a broad analysis without needing to manually locate individual event log files. This approach is ideal for large-scale evaluation of NQE suitability.

• Qualification Script (Alternative): This is an alternative method for advanced or specific use cases. It is a shell script that analyzes a single Spark event log file or all event logs within a specific Cloud Storage directory. Use this method if you have the Cloud Storage path to the event logs you want to analyze.

Qualification tool output files

Once the qualification job or script analysis is complete, the qualification tool places the following output files in a perfboost-output directory in the current directory:

• AppsRecommendedForBoost.tsv: A tab-separated list of applications recommended for use with Native Query Execution.

• UnsupportedOperators.tsv: A tab-separated list of applications not recommended for use with Native Query Execution.

AppsRecommendedForBoost.tsv output file

The following table shows the contents of a sample AppsRecommendedForBoost.tsv output file. It contains a row for each analysed application.

Sample AppsRecommendedForBoost.tsv output file:

Column descriptions:

• applicationId: The ApplicationID of the Spark application. Use this to identify the corresponding batch workload.

• applicationName: The name of the Spark application.

• rddPercentage: The percentage of RDD operations in the application. RDD operations are not supported by Native Query Execution.

• unsupportedSqlPercentage: Percentage of SQL operations not supported by Native Query Execution.

• totalTaskTime: Cumulative task time of all tasks executed during the application run.

• supportedTaskTime: The total task time supported by Native Query Execution.

The following columns provide important information to help you determine if Native Query Execution can benefit your batch workload:

• supportedSqlPercentage: The percentage of SQL operations supported by native query execution. The higher the percentage, the greater the runtime reduction that can be achieved by running the application with Native Query Execution.

• recommendedForBoost: If TRUE, running the application with Native Query Execution is recommended. If recommendedForBoost is FALSE, don't use Native Query Execution on the batch workload.

• expectedRuntimeReduction: The expected percentage reduction in application runtime when you run the application with Native Query Execution.

UnsupportedOperators.tsv output file.

The UnsupportedOperators.tsv output file contains a list of operators used in workload applications that are not supported by Native Query Execution. Each row in the output file lists an unsupported operator.

Column descriptions:

• unsupportedOperator: The name of the operator that is not supported by Native Query Execution.

• cumulativeCpuMs: The number of CPU milliseconds consumed during the execution of the operator. This value reflects the relative importance of the operator in the application.

• count: The number of times the operator is used in the application.

Use Native Query Execution

You can use Native Query Execution with you application by setting Native Query Execution properties when you create the batch workload, interactive session, or session template that runs you application.

Use Native Query Execution with batch workloads

You can use the Google Cloud console, Google Cloud CLI, or Dataproc API to enable Native Query Execution on a batch workload.

gcloud dataproc batches submit spark \
    --project=PROJECT_ID \
    --region=REGION \
    --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
    --class=org.apache.spark.examples.SparkPi \
    --properties=spark.dataproc.runtimeEngine=native,spark.dataproc.driver.compute.tier=premium,spark.dataproc.executor.compute.tier=premium \
    OTHER_FLAGS_AS_NEEDED

When to use Native Query Execution

Use Native Query Execution in the following scenarios:

• Spark Dataframe APIs, Spark Dataset APIs, and Spark SQL queries that read data from Parquet and ORC files. The output file format doesn't affect Native Query Execution performance.

• Workloads recommended by the Native Query Execution qualification tool.

When not to use Native Query Execution

Inputs of the following data types:

• Byte: ORC and Parquet
• Timestamp: ORC
• Struct, Array, Map: Parquet

Limitations

Enabling Native Query Execution in the following scenarios can cause exceptions, Spark incompatibilities, or workload fallback to the default Spark engine.

Fallbacks

Native Query Execution in the following the execution can result in workload fallback to the Spark execution engine, resulting in regression or failure.

• ANSI: If ANSI mode is enabled, execution falls back to Spark.

• Case-sensitive mode: Native Query Execution supports the Spark default case-insensitive mode only. If case-sensitive mode is enabled, incorrect results can occur.

• Partitioned table scan: Native Query Execution supports the partitioned table scan only when the path contains the partition information, otherwise the workload falls back to the Spark execution engine.

Incompatible behavior

Incompatible behavior or incorrect results can result when using Native query execution in the following cases:

• JSON functions: Native Query Execution supports strings surrounded by double quotes, not single quotes. Incorrect results occur with single quotes. Using "*" in the path with the get_json_object function returns NULL.

• Parquet read configuration:

  • Native Query Execution treats spark.files.ignoreCorruptFiles as set to the default false value, even when set to true.
  • Native Query Execution ignores spark.sql.parquet.datetimeRebaseModeInRead, and returns only the Parquet file contents. Differences between the legacy hybrid (Julian Gregorian) calendar and the Proleptic Gregorian calendar are not considered. Spark results can differ.

• NaN: Not supported. Unexpected results can occur, for example, when using NaN in a numeric comparison.

• Spark columnar reading: A fatal error can occur due since the Spark columnar vector is incompatible with Native Query Execution.

• Spill: When shuffle partitions are set to a large number, the spill-to-disk feature can trigger an OutOfMemoryException. If this occurs, reducing the number of partitions can eliminate this exception.