UDF Connection to Lakehouse does not work

[michael_muell]

Hi all, 

I'm trying to get a UDF to work that creates a file in the lakehouse. I saw that there is a sample that almost achieves what I want. Unfortunately it does not work when I connect my lakehouse. I get the following error: 

Running another sample snippet connecting with a warehouse item works perfectly fine. It seems to be some problem specifically with the lakehouse connection.

Can anyone help?  

{
"functionName": "write_csv_file_in_lakehouse",
"invocationId": "00000000-0000-0000-0000-000000000000",
"status": "Failed",
"errors": [
{
"errorCode": "WorkloadException",
"subErrorCode": "NotFound",
"message": "User data function: 'write_csv_file_in_lakehouse' invocation failed."
}
]
}

The connection to the datasource is setup: 

This is the code I'm running: 

import datetime

import fabric.functions as fn

import logging

udf = fn.UserDataFunctions()

import pandas as pd

import datetime

# Select 'Manage connections' and add a connection to a Lakehouse.

#Replace the alias "<My Lakehouse alias>" with your connection alias.

@udf.connection(argName="myLakehouse", alias="dateiupload")

@udf.function()

def write_csv_file_in_lakehouse(myLakehouse: fn.FabricLakehouseClient, employees: list)-> str:

    logging.info('test')

    '''

    Description: Write employee data to lakehouse as timestamped CSV file using pandas.

    Args:

        myLakehouse (fn.FabricLakehouseClient): Fabric lakehouse connection.

        employees (list): List of employee records as [ID, Name, DeptID] arrays.

    Returns:

        str: Confirmation message with filename and viewing instructions.

    Example:

        employees = [[1,"John Smith", 31], [2,"Kayla Jones", 33]]

        Creates "Employees1672531200.csv" in lakehouse

    '''

    csvFileName = "Employees" + str(round(datetime.datetime.now().timestamp())) + ".csv"

    # Convert the data to a DataFrame

    df = pd.DataFrame(employees, columns=['ID','EmpName', 'DepID'])

    # Write the DataFrame to a CSV file

    csv_string = df.to_csv(index=False)

    # Upload the CSV file to the Lakehouse

    connection = myLakehouse.connectToFiles()

    csvFile = connection.get_file_client(csvFileName)  

    csvFile.upload_data(csv_string, overwrite=True)

    csvFile.close()

    connection.close()

    return f"File {csvFileName} was written to the Lakehouse. Open the Lakehouse in https://app.fabric.microsoft.com to view the files"

[v-venuppu · Accepted solution]

Hi @michael_muell ,

Please use the below code:

import pandas as pd

import datetime

import fabric.functions as fn

import logging

udf = fn.UserDataFunctions()

@udf.connection(argName="myLakehouse", alias="LH123")

@udf.function()

def write_csv_file_in_lakehouse(myLakehouse: fn.FabricLakehouseClient, employees: list) -> str:

    """

    Writes employee data to Lakehouse Files as a CSV.

    """

    logging.info("Starting CSV file write to Lakehouse")

    # Create timestamped filename

    csvFileName = "Employees_" + str(round(datetime.datetime.now().timestamp())) + ".csv"

    # Create DataFrame and CSV string

    df = pd.DataFrame(employees, columns=["ID", "EmpName", "DepID"])

    csv_string = df.to_csv(index=False)

    csv_bytes = csv_string.encode("utf-8")  # Convert string to bytes

    # Connect to Lakehouse Files and upload

    connection = myLakehouse.connectToFiles()

    file_client = connection.get_file_client(csvFileName)

    file_client.upload_data(csv_bytes, overwrite=True)

    # Close connections

    file_client.close()

    connection.close()

    return f"File '{csvFileName}' was uploaded successfully."

Add Pandas library as shown below:

To test this I have created pipeline and it worked for me:

File got created in Lakehouse as shown below:

Thank you.

[michael_muell]

This works! Thanks a lot! 

[BhaveshPatel]

This is how it works:
Python (Pandas ) -- > Apache Spark ( Data Lake ) --> Delta Lake ( Data Lakehouse ) 

  # Convert the data to a DataFrame

    df = pd.DataFrame(employees, columns=['ID','EmpName', 'DepID'])

# Covert DataFrame to Data Lake

sdf = spark.createDataFrame(df)

or 

sdf = spark.read.csv("Employees_" + str(round(datetime.datetime.now().timestamp())) + ".csv",header = True)

# Writing a DataFrame to Delta format

sdf.write.format("delta").option("overwrite").saveAsTable("DimEmployees")

You are missing Spark Dataframe ( Data Lake ).

[v-venuppu]

Hi @michael_muell ,

Thank you for reaching out to Microsoft Fabric Community.

The issue is likely caused by a mismatch in the Lakehouse connection alias or trying to upload string data using the wrong method. Make sure your UDF connection alias (dateiupload) matches exactly what’s set in the UDF UI. Also, replace upload_data with upload_text since you're uploading a CSV string, not binary data. Here's the fix:
csvFile.upload_text(csv_string, overwrite=True)

Here's a documentation for your reference:
fabric.functions.FabricLakehouseClient class | Microsoft Learn

Thank you.

[michael_muell]

Hi @v-venuppu 
Thanks for the reply. I triple checked and the alias is exactly matching. 
Also the code change did not help. The problem is the connection to the lakehouse. 

Any other suggestions?

Best regards 
Michael