Databricks Databricks-Certified-Associate-Developer-for-Apache-Spark-3.5 - Databricks Certified Associate Developer for Apache Spark 3.5-Python

Comprehensive and Detailed Explanation From Exact Extract:

Option B is the correct and idiomatic approach in PySpark to split a string column (like email) based on a delimiter such as "@".

The split(col("email"), "@") function returns an array with two elements: username and domain.

getItem(0) retrieves the first part (username).

getItem(1) retrieves the second part (domain).

withColumn() is used to create new columns from the extracted values.

Example from official Databricks Spark documentation on splitting columns:

from pyspark.sql.functions import split, col

df.withColumn("username", split(col("email"), "@").getItem(0)) \

withColumn("domain", split(col("email"), "@").getItem(1))

⚠️Why other options are incorrect:

A uses fixed substring indices (substr(0, 5)), which won’t correctly extract usernames and domains of varying lengths.

C uses substring_index, which is available but less idiomatic for splitting emails and is slightly less readable.

D removes "@" from the email entirely, losing the separation between username and domain, and ends up duplicating values in both fields.

Therefore, Option B is the most accurate and reliable solution according to Apache Spark 3.5 best practices.

Comprehensive and Detailed Explanation From Exact Extract:

The correct way to load specific columns from an ORC file is to first load the file using.load()and then apply.select()on the resulting DataFrame. This is valid with.read.format("orc")or the shortcut.read.orc().

df = spark.read.format("orc").load("/file/test_data.orc").select("col1","col2")

Why others are incorrect:

Aperforms selection after filtering, but doesn’t match the intention to minimize memory at load.

Bincorrectly tries to use.select()before.load(), which is invalid.

Cuses a non-existent.selected()method.

Dcorrectly loads and then selects.

The most efficient and correct approach is to define a schema using StructType and pass it tospark.read.schema(...).

This avoids schema inference overhead and ensures proper data types are enforced during read.

Example:

frompyspark.sql.typesimportStructType, StructField, StringType, DoubleType

schema = StructType([

StructField("order_id", StringType(),True),

StructField("amount", DoubleType(),True),

])

df = spark.read.schema(schema).json("path/to/json")

— Source:Databricks Guide – Read JSON with predefined schema

Comprehensive and Detailed Explanation From Exact Extract:

The correct output mode for streaming aggregations that need to output the full updated results at each trigger is"complete".

From the official documentation:

“complete: The entire updated result table will be output to the sink every time there is a trigger.”

This is ideal for aggregations, such as counts or averages grouped by a key, where the result table changes incrementally over time.

append: only outputs newly added rows

replace and aggregate: invalid values for output mode

Comprehensive and Detailed Explanation From Exact Extract:

When converting a standard PySpark UDF to a Pandas UDF for performance optimization, the function must operate on a Pandas Series as input and return a Pandas Series as output.

In this case, the original function signature:

def shake_256(raw: str) -> str

is scalar — not compatible with Pandas UDFs.

According to the official Spark documentation:

“Pandas UDFs operate onpandas.Seriesand returnpandas.Series. The function definition should be:

def my_udf(s: pd.Series) -> pd.Series:

and it must be registered usingpandas_udf(...).”

Therefore, to fix the error:

The function should be updated to:

def shake_256(df: pd.Series) -> pd.Series:

return df.apply(lambda x: hashlib.shake_256(x.encode()).hexdigest(20))

This will allow Spark to efficiently execute the Pandas UDF in vectorized form, improving performance compared to standard UDFs.

dedup_df = iot_bronze_df.dropDuplicates(["event_ts","sensor_id","metric_value"])

dropDuplicates accepts a list of columns to use for deduplication.

This ensures only unique records based on the specified keys are retained.

Comprehensive and Detailed Explanation From Exact Extract:

df.cache()is shorthand fordf.persist(StorageLevel.MEMORY_AND_DISK)

df.persist()allows specifying any storage level such asMEMORY_ONLY,DISK_ONLY,MEMORY_AND_DISK_SER, etc.

By default,persist()usesMEMORY_AND_DISK, unless specified otherwise.

f.to_date() converts a timestamp or string to a DateType.

Ideal for extracting the date component (year-month-day) from a full timestamp.

Example:

frompyspark.sql.functionsimportto_date

dates_df.withColumn("date", to_date("timestamp"))

Comprehensive and Detailed Explanation From Exact Extract:

If each of the 10 nodes runs 4 executors, and each executor is assigned 4 CPU cores:

Executors per node = 4

Cores per executor = 4

Total executors = 4 * 10 = 40

Total cores = 40 executors * 4 cores = 160 cores

However, Spark uses 1 core for overhead on each node when managing multiple executors. Thus, the practical allocation is:

Total usable executors = 4 executors/node × 10 nodes = 40

Total cores = 4 cores × 40 executors = 160

Answer: A — but the question asks specifically about “CPU cores used by the application,” assuming all allocated cores are usable (as Spark typically runs executors without internal core reservation unless explicitly configured).

However, if you are considering 4 executors/node × 4 cores = 16 cores per node, across 10 nodes, that’s 160.

Final Answer: A