I’m working with a large dataframe in Databricks, processing it in a streaming-batch fashion (I’m reading as a stream, but using .trigger(availableNow=True) for batch-like processing).

I’m fetching around 40GB of CDF updates daily and performing some heavy aggregations in foreachBatch mode.

spark.readStream.format("delta")
    .option("readChangeFeed", "true")
    .option("maxBytesPerTrigger", "10G")


def process_and_upsert(df, batch_id):
    # heavy aggs and upsert:
    # pass

df_streaming_components.writeStream.format(DeltaUtils.FORMAT_DELTA)
    .option("checkpointLocation", checkpoint_loc)
    .foreachBatch(process_and_upsert)
    .outputMode("update")
    .trigger(availableNow=True)
    .start()
    .awaitTermination()

The tables are partitioned based on some key, e.g. class_number. Could I use that to read the CDF changes in partitions? For instance, I would like to process and stream changes by getting class_number="class_a"changes process and upsert. Next changes_df for "class_b" process and upsert ... Since my aggregations are heavy, it would be far quicker if I can get changes based on my partitioned column.