Spark JDBC, saveAsTable and Temp Views

1.Reading and writing databases over JDBC

The reader and writer that handle files also connect to relational databases. spark.read.jdbc(url, table, properties) returns a DataFrame for a database table, and df.write.jdbc(url, table, mode, properties) writes one back. On a read, "Spark automatically reads the schema from the external database table and maps its types to Spark SQL types."

To read efficiently, add four arguments to the read: a numeric column, its lowerBound and upperBound, and numPartitions. Spark then splits the read into that many partitions so it can load them in parallel. Without these arguments, nothing is set up for a parallel read.

# Read with partitioning for parallel loading
df = spark.read.jdbc(
    url="jdbc:postgresql://localhost:5432/mydb",
    table="users",
    column="id",
    lowerBound=1,
    upperBound=1000,
    numPartitions=10,
    properties={"user": "myuser", "password": "mypassword"}
)

You can also use the generic format(...) / option(...) form instead of jdbc(). Instead of a table, you pass a native SQL statement through the query option. According to the documentation, this "ensures that filter and join logic executes on the source database before data reaches Spark." Spark doesn't do transformation pushdown on the query string. Whatever you put in it runs as written on the source, and later DataFrame operations run on the Spark cluster.

df = (spark.read
  .format("sqlserver")
  .option("host", "<your-sql-server-instance>.database.windows.net")
  .option("user", dbutils.secrets.get(scope="<scope>", key="<user>"))
  .option("password", dbutils.secrets.get(scope="<scope>", key="<password>"))
  .option("database", "<database-name>")
  .option("query", "SELECT id, name FROM users WHERE active = 1")
  .load())

For JDBC writes, you choose the mode the same way as for files: "Use append to add rows to an existing table or overwrite to replace its contents."

# Write to database table
df.write.jdbc(
    url="jdbc:postgresql://localhost:5432/mydb",
    table="users",
    mode="overwrite",
    properties={"user": "myuser", "password": "mypassword"}
)

Sources12

2.saveAsTable, insertInto, bucketBy and sortBy

Writing to a path gives you files. saveAsTable(name, format, mode, partitionBy, **options) instead "Saves the content of the DataFrame as the specified table", so you can query it by name later. It uses the same mode() values as a file write, and partitionBy works on it too. The reference shows df.write.mode("overwrite").format("parquet").partitionBy("year").saveAsTable("partitioned_table"). To load rows into a table that already exists, use insertInto(tableName, overwrite). Passing overwrite=True replaces the table's contents instead of appending.

Two more writer methods control layout and only take effect on tables. bucketBy(numBuckets, col, ...) hashes rows into a fixed number of buckets. sortBy(col, ...) "Sorts the output in each bucket by the given columns on the file system." The bucketBy reference says these methods are "Applicable for file-based data sources in combination with DataFrameWriter.saveAsTable." sortBy sorts inside buckets, so you use it together with bucketBy.

# Save as managed table with options
df.write \
    .mode("overwrite") \
    .format("parquet") \
    .partitionBy("year") \
    .saveAsTable("partitioned_table")

Sources345

3.Temporary views: querying a DataFrame with Spark SQL

Data you have read doesn't need to be written anywhere before SQL can query it. createOrReplaceTempView(name) registers the DataFrame under a name, and spark.sql("SELECT ... FROM name") queries it. A temp view stores no data. It is a name for the DataFrame, and "The lifetime of this temporary table is tied to the SparkSession that was used to create this DataFrame."

The related methods differ in two ways: what happens if the name is already taken, and how long the view lasts. createTempView refuses to replace an existing view and throws TempTableAlreadyExistsException. The createOrReplace... versions replace it without an error. A global temp view is tied to the whole Spark application instead of one session, and you read it through the global_temp schema, as in spark.table("global_temp.people"). Databricks notes that global_temp views aren't supported on serverless compute and recommends createOrReplaceTempView there.

df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"])
df.createOrReplaceTempView("people")

df2 = df.filter(df.age > 3)
df2.createOrReplaceTempView("people")
df3 = spark.sql("SELECT * FROM people")
assert sorted(df3.collect()) == sorted(df2.collect())

Sources67

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Spark optimises the SQL you pass in the query option and pushes only part of it to the database.Why is that wrong?

   The query string runs entirely on the external source, and Spark fetches the result without pushing its own transformations into it.

   Covered in Reading and writing databases over JDBC

2. 2.bucketBy and sortBy work on any file write, such as df.write.bucketBy(...).parquet(path).Why is that wrong?

   Bucketing is documented for file-based sources together with saveAsTable, and sortBy sorts within those buckets.

   Covered in saveAsTable, insertInto, bucketBy and sortBy

3. 3.A view from createOrReplaceGlobalTempView can be queried by its bare name, just like a session temp view.Why is that wrong?

   Global temp views belong to the Spark application and are read through the global_temp schema, for example spark.table("global_temp.people").

   Covered in Temporary views: querying a DataFrame with Spark SQL

Practise it for real

Write a bucketed, sorted table with saveAsTable, read it back by name, then clean it up.

1. 1.Run spark.sql("DROP TABLE IF EXISTS sorted_bucketed_table")

   Why: Lets you rerun the exercise without the write hitting an existing table

   You should see: The statement finishes without error whether or not the table existed

2. 2.Create a DataFrame of (100, "Alice"), (120, "Alice"), (140, "Bob") with schema ["age", "name"] and write it with .write.bucketBy(1, "name").sortBy("age").mode("overwrite").saveAsTable("sorted_bucketed_table")

   Why: bucketBy and sortBy take effect together with saveAsTable

   You should see: A table named sorted_bucketed_table now exists

3. 3.Run spark.read.table("sorted_bucketed_table").sort("age").show()

   Why: Confirms the table can be read back by name

   You should see: Three rows: 100 Alice, 120 Alice, 140 Bob

4. 4.Run spark.sql("DROP TABLE sorted_bucketed_table")

   Why: Removes the table you created

   You should see: The table no longer exists