Forum Discussion
Reusable function for data transformation - user data functions
HI v-ssriganesh ,
As I explained in my previous response, based on Microsoft's response, the User data Functions cannot be used for transformations in the dataframe. Therefore, we need to utilize PySpark's native functions to transform data in the User data functions. So, I need to tweak my solution a bit and not use User data functions for Fabric but instead use pyspark.udf to do the transformation.
I think I know the way ahead now. Thanks for your support and help. We can close the ticket now.
Hi tinbaj,
Thank you for sharing the details and code. The error TypeError: Column is not iterable occurs because the User Data Function (UDF) is being applied to a Spark DataFrame column directly, which isn't compatible with the function's expectation of a single string input. To fix this, you need to register the UDF with Spark to handle DataFrame columns.
Here’s how to resolve it:
In your notebook, after instantiating the UDF, register it with Spark:
- Use spark.udf.register to make the UDF available for DataFrame operations.
- Then, apply it using withColumn with the registered UDF.
Update your notebook code as follows:
- Instantiate the UDF: data_functions = notebookutils.udf.getFunctions('data_functions')
- Register the UDF: spark.udf.register("convert_julian_to_date", data_functions.convert_julian_to_date)
- Apply to the DataFrame: df_silver = df_silver.withColumn('request_date', spark.sql.functions.expr("convert_julian_to_date(request_date)"))
This ensures the UDF processes each row’s request_date column value correctly. Also, verify that the request_date column in df_silver contains valid Julian date strings (e.g: '123241'). If the column has mixed or invalid data types, you may need to preprocess it to ensure all values are strings.
If this helps, please mark it as “Accept as solution” and feel free to give a “Kudos” to help others in the community as well.
Thank you.
Hi v-ssriganesh ,
Thanks for your response. I am getting this error when I am running the command to register the UDF: Register the UDF: spark.udf.register("convert_julian_to_date", data_functions.convert_julian_to_date)
--> 612 self.sparkSession._jsparkSession.udf().registerPython(name, register_udf._judf) 613 return return_udf File /opt/spark/python/lib/pyspark.zip/pyspark/sql/udf.py:321, in UserDefinedFunction._judf(self) 314 @property 315 def _judf(self) -> JavaObject: 316 # It is possible that concurrent access, to newly created UDF, 317 # will initialize multiple UserDefinedPythonFunctions. 318 # This is unlikely, doesn't affect correctness, 319 # and should have a minimal performance impact. 320 if self._judf_placeholder is None: --> 321 self._judf_placeholder = self._create_judf(self.func) 322 return self._judf_placeholder File /opt/spark/python/lib/pyspark.zip/pyspark/sql/udf.py:330, in UserDefinedFunction._create_judf(self, func) 327 spark = SparkSession._getActiveSessionOrCreate() 328 sc = spark.sparkContext --> 330 wrapped_func = _wrap_function(sc, func, self.returnType) 331 jdt = spark._jsparkSession.parseDataType(self.returnType.json()) 332 assert sc._jvm is not None File /opt/spark/python/lib/pyspark.zip/pyspark/sql/udf.py:59, in _wrap_function(sc, func, returnType) 57 else: 58 command = (func, returnType) ---> 59 pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command) 60 assert sc._jvm is not None 61 return sc._jvm.SimplePythonFunction( 62 bytearray(pickled_command), 63 env, (...) 68 sc._javaAccumulator, 69 ) File /opt/spark/python/lib/pyspark.zip/pyspark/rdd.py:5251, in _prepare_for_python_RDD(sc, command) 5248 def _prepare_for_python_RDD(sc: "SparkContext", command: Any) -> Tuple[bytes, Any, Any, Any]: 5249 # the serialized command will be compressed by broadcast 5250 ser = CloudPickleSerializer() -> 5251 pickled_command = ser.dumps(command) 5252 assert sc._jvm is not None 5253 if len(pickled_command) > sc._jvm.PythonUtils.getBroadcastThreshold(sc._jsc): # Default 1M 5254 # The broadcast will have same life cycle as created PythonRDD File /opt/spark/python/lib/pyspark.zip/pyspark/serializers.py:469, in CloudPickleSerializer.dumps(self, obj) 467 msg = "Could not serialize object: %s: %s" % (e.__class__.__name__, emsg) 468 print_exec(sys.stderr) --> 469 raise pickle.PicklingError(msg) PicklingError: Could not serialize object: PySparkRuntimeError: [CONTEXT_ONLY_VALID_ON_DRIVER] It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.
Hi tinbaj,
Thank you for providing the error details. The PicklingError: [CONTEXT_ONLY_VALID_ON_DRIVER] occurs because the User Data Function (UDF) is being serialized in a way that references the SparkContext, which isn't allowed in Spark's distributed environment. This is likely due to how the UDF is defined or accessed in your notebook.To resolve this, try the following steps:
Instead of directly registering the UDF with spark.udf.register, use the Fabric UDF directly in the DataFrame operation, as Fabric’s UDFs are designed to work seamlessly with Spark. Update your notebook code as follows:
- Instantiate the UDF: data_functions = notebookutils.udf.getFunctions('data_functions')
- Apply the UDF to the DataFrame: df_silver = df_silver.withColumn('request_date', data_functions.convert_julian_to_date(df_silver.request_date))
- Ensure your UDF (convert_julian_to_date) in the User Data Functions item doesn’t reference SparkContext or other non-serializable objects. Your provided UDF code looks fine, but confirm it only uses standard Python libraries (e.g., datetime, timedelta) and avoids Spark-specific calls.
- Before applying to the DataFrame, test the UDF with a single value to confirm it works: print(data_functions.convert_julian_to_date('123241')). This should return '2023-08-29 00:00:00'.
If the error persists, please share:
- The schema of df_silver (df_silver.printSchema()).
- Any modifications made to the UDF code.
- Whether you’re running this in a Fabric notebook with a Spark session active.
Please try these steps and let me know the outcome. If it resolves the issue, consider marking it as “Accept as solution” and giving a “Kudos” to help others in the community.
Thank you.Hi v-ssriganesh ,
Thanks for your response, but the suggested code did not fix the problem. I can confirm that I am using standard python libraries in UDF and does not use spark context.
The implementation as per the suggestion and the error message is as below:
data_functions = notebookutils.udf.getFunctions('data_functions')print(data_functions.convert_julian_to_date('123241'))Return Value: 2023-08-29df_silver = df_silver.withColumn("request_date", data_functions.convert_julian_to_date(df_silver.request_date))Error Message: PySparkTypeError: [NOT_ITERABLE] Column is not iterable.Now if we go through the documentation for UDF's (Link: https://learn.microsoft.com/en-us/fabric/data-engineering/user-data-functions/python-programming-model), column data type is not one of the acceptable data type in UDF's. could this be a reason for this error?ThanksHello tinbaj,
Thank you for the update and detailed feedback.
he PySparkTypeError: [NOT_ITERABLE] Column is not iterable error occurs because Fabric User Data Functions (UDFs) expect scalar inputs (e.g:L strings, integers), but df_silver.request_date is a Spark DataFrame column, which isn’t directly compatible. The documentation you referenced correctly notes that UDFs don’t accept column objects as inputs, which explains this error.To resolve this, you need to register the UDF with Spark to process each row’s request_date value individually. Since you’ve confirmed the UDF works for a single input ('123241' returns '2023-08-29'), the issue is specific to DataFrame application. Here’s how to fix it:
- Instantiate the UDF: data_functions = notebookutils.udf.getFunctions('data_functions')
- Register the UDF with Spark: Use from pyspark.sql.functions import udf and register the UDF as convert_udf = udf(data_functions.convert_julian_to_date).
- Apply the UDF to the DataFrame: df_silver = df_silver.withColumn('request_date', convert_udf(df_silver.request_date)).
Additionally, check the request_date column is a string type, as your UDF expects strings.
If this helps, please “Accept as solution” and give a “kudos” to assist other community members.
Thank you.