PySpark UDF - read and write other dataframe

within a UDF I want to read a delta table into a dataframe, based on its content update the row of the actual dataframe, on which the UDF is applied, and then update the delta table. I would use the UDF within a structured streaming foreachbatch. How is this possible?

df_other = spark.read.format("delta").load(path)

@udf(StringType())
def my_udf(df_other: DataFrame) -> str:
    ...
    # some things to do based on df_other's content.
    ...
    df_new_rows = ...
    df_new_rows.write.format("delta").mode("append").write(path)
    ...
    return "wathever"

Use the following UDF to read and update the delta table:

def read_and_process_delta_table(spark, table_path):
    # Define UDF to perform operations on DataFrame
    @udf(StringType())
    def my_combined_udf(name, age):
        # Perform operations based on name and age
        # Example: Concatenate name and age
        return f"{name}_{age}"

    # Read Delta table
    delta_df = spark.read.format("delta").load(table_path)

    # Apply combined UDF to DataFrame
    processed_df = delta_df.withColumn("processed_column", my_combined_udf(delta_df["name"], delta_df["age"]))
    processed_df.write.format("delta").mode("overwrite").option("mergeSchema", "true").save(table_path)
    return processed_df

It will read and update the dataframe successfully, as shown below:

Delta table:

Updated delta table:

Here is the complete code for your reference:

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# Function to read Delta table
def read_and_process_delta_table(spark, table_path):
    # Define UDF to perform operations on DataFrame
    @udf(StringType())
    def my_combined_udf(name, age):
        # Perform operations based on name and age
        # Example: Concatenate name and age
        return f"{name}_{age}"

    # Read Delta table
    delta_df = spark.read.format("delta").load(table_path)

    # Apply combined UDF to DataFrame
    processed_df = delta_df.withColumn("processed_column", my_combined_udf(delta_df["name"], delta_df["age"]))
    processed_df.write.format("delta").mode("overwrite").option("mergeSchema", "true").save(table_path)
    return processed_df

# Usage
spark = SparkSession.builder.appName("DeltaUDFExample").getOrCreate()
table_path = "/mnt/files/delta_table"

# Read and process Delta table
result = read_and_process_delta_table(spark, table_path)

# Show the result
result.show()

You can refer this to UDF within a structured streaming.