Partekatu bidez

Introducción al modelo de programación de funciones de datos de usuario de Fabric

El modelo de programación de funciones de datos de usuario de Fabric define los patrones y conceptos para crear funciones en Fabric.

El fabric-user-data-functions SDK implementa este modelo de programación, lo que proporciona la funcionalidad necesaria para crear y publicar funciones ejecutables. El SDK también le permitirá integrarse sin problemas con otros elementos del ecosistema de Fabric, como orígenes de datos de Fabric. Esta biblioteca está disponible públicamente en PyPI y se encuentra preinstalada en los elementos de funciones de datos de usuario.

En este artículo se explica cómo usar el SDK para compilar funciones que se pueden invocar desde el portal de Fabric, otros elementos de Fabric o aplicaciones externas mediante la API rest. Aprenderá el modelo de programación y los conceptos clave con ejemplos prácticos.

Sugerencia

Para obtener información completa sobre todas las clases, métodos y parámetros, consulte la documentación de referencia del SDK.

Introducción al SDK

En esta sección se presentan los componentes principales del SDK de User Data Functions y se explica cómo estructurar las funciones. Aprenderá sobre las importaciones requeridas, los decoradores y los tipos de datos de entrada y salida que sus funciones pueden manejar.

SDK de funciones de datos de usuario

El fabric-user-data-functions SDK proporciona los componentes principales que necesita para crear funciones de datos de usuario en Python.

Importaciones e inicialización necesarias

Cada archivo de funciones de datos de usuario debe importar el fabric.functions módulo e inicializar el contexto de ejecución:

import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()

El @udf.function() decorador

Las funciones marcadas con el @udf.function() decorador se pueden invocar desde el portal de Fabric, otro elemento de Fabric o una aplicación externa. Las funciones con este decorador deben especificar un tipo de valor devuelto.

Ejemplo:

@udf.function()
def hello_fabric(name: str) -> str:
    logging.info('Python UDF trigger function processed a request.')
    logging.info('Executing hello fabric function.')
    
    return f"Welcome to Fabric Functions, {name}, at {datetime.datetime.now()}!"

Funciones del asistente

Los métodos de Python sin el @udf.function() decorador no se pueden invocar directamente. Solo se pueden llamar desde funciones decoradas y actúan como funciones auxiliares.

Ejemplo:

def uppercase_name(name: str) -> str:
    return name.upper()

Tipos de entrada admitidos

Es posible definir parámetros de entrada para la función, como tipos de datos primitivos como str, int, float, etc. Los tipos de datos de entrada admitidos son:

Tipo JSON Tipo de datos de Python
Cuerda Str
Cadena Fecha y hora fecha y hora
Booleano booleano
Números entero, flotante
Matriz list[], ejemplo list[int]
objeto diccionario
objeto DataFrame de Pandas
Objeto o matriz de objetos Serie pandas

Nota:

Para usar los tipos DataFrame y Series de Pandas, vaya al portal de Fabric, busque el área de trabajo y abra el elemento de funciones de datos de usuario. Seleccione Administración de bibliotecas, busque el fabric-user-data-functions paquete y actualícelo a la versión 1.0.0 o posterior.

Ejemplo de cuerpo de solicitud para los tipos de entrada admitidos:

{
  "name": "Alice",                          // String (str)
  "signup_date": "2025-11-08T13:44:40Z",    // Datetime string (datetime)
  "is_active": true,                        // Boolean (bool)
  "age": 30,                                // Number (int)
  "height": 5.6,                            // Number (float)
  "favorite_numbers": [3, 7, 42],           // Array (list[int])
  "profile": {                              // Object (dict)
    "email": "alice@example.com",
    "location": "Sammamish"
  },
  "sales_data": {                           // Object (pandas DataFrame)
    "2025-11-01": {"product": "A", "units": 10},
    "2025-11-02": {"product": "B", "units": 15}
  },
  "weekly_scores": [                        // Object or Array of Objects (pandas Series)
    {"week": 1, "score": 88},
    {"week": 2, "score": 92},
    {"week": 3, "score": 85}
  ]
}

Tipos de salida compatibles

Los tipos de datos de salida admitidos son:

Tipo de datos de Python
Str
fecha y hora
booleano
entero, flotante
list[data-type], por ejemplo list[int]
diccionario
Ninguno
Serie pandas
DataFrame de Pandas

Escritura de funciones

Requisitos y limitaciones de sintaxis

Al escribir Funciones de datos de usuario, debe seguir reglas de sintaxis específicas para asegurarse de que las funciones funcionan correctamente.

Nomenclatura de parámetros

  • Usar camelCase: los nombres de parámetro deben usar la convención de nomenclatura camelCase y no pueden contener caracteres de subrayado. Por ejemplo, usa productName en lugar de product_name.
  • Palabras clave reservadas: no se pueden usar palabras clave reservadas de Python ni las siguientes palabras clave específicas de Fabric como nombres de parámetro o nombres de función: req, contexty reqInvocationId.

Requisitos de parámetros

  • No se admiten valores predeterminados: no se admiten los valores de parámetro predeterminados. Todos los parámetros son necesarios al invocar una función. Por ejemplo, la función siguiente produce un error de sintaxis: 
    # The default value for the argument called 'name' is not supported and treated like a syntax error.
@udf.function()
def goodbye_fabric(name: str = "N/A") -> str:
    return f"Goodbye, {name}."
  • Anotaciones de tipo necesarias: todos los parámetros deben incluir anotaciones de tipo (por ejemplo, name: str).

Requisitos de función

  • Tipo de valor devuelto necesario: las funciones con el @udf.function() decorador deben especificar una anotación de tipo de valor devuelto (por ejemplo, -> str).
  • Importaciones necesarias: la declaración y import fabric.functions as fn la udf = fn.UserDataFunctions() inicialización son necesarias para que tus funciones funcionen correctamente.

Ejemplo de sintaxis correcta

@udf.function()
def process_order(orderNumber: int, customerName: str, orderDate: str) -> dict:
    return {
        "order_id": orderNumber,
        "customer": customerName,
        "date": orderDate,
        "status": "processed"
    }

Cómo escribir una función asincrónica

Agregue un decorador asincrónico con la definición de función en el código. Con una async función, puede mejorar la capacidad de respuesta y la eficacia de la aplicación mediante el control de varias tareas a la vez. Son ideales para administrar grandes volúmenes de operaciones dependientes de E/S. Esta función de ejemplo lee un archivo CSV desde un almacén de datos en lagos utilizando pandas. La función toma el nombre de archivo como parámetro de entrada.

import pandas as pd 

# Replace the alias "<My Lakehouse alias>" with your connection alias.
@udf.connection(argName="myLakehouse", alias="<My Lakehouse alias>")
@udf.function()
async def read_csv_from_lakehouse(myLakehouse: fn.FabricLakehouseClient, csvFileName: str) -> str:

    # Connect to the Lakehouse
    connection = myLakehouse.connectToFilesAsync()   

    # Download the CSV file from the Lakehouse
    csvFile = connection.get_file_client(csvFileName)

    downloadFile = await csvFile.download_file()
    csvData = await downloadFile.readall()
    
    # Read the CSV data into a pandas DataFrame
    from io import StringIO
    df = pd.read_csv(StringIO(csvData.decode('utf-8')))

    # Display the DataFrame    
    result="" 
    for index, row in df.iterrows():
        result=result + "["+ (",".join([str(item) for item in row]))+"]"
    
    # Close the connection
    csvFile.close()
    connection.close()

    return f"CSV file read successfully.{result}"

Trabajar con datos

Conexiones de datos a orígenes de datos de Fabric

El SDK permite hacer referencia a conexiones de datos sin necesidad de escribir cadenas de conexión en el código. La biblioteca fabric.functions proporciona dos maneras de controlar las conexiones de datos:

  • fabric.functions.FabricSqlConnection: permite trabajar con bases de datos SQL en Fabric, incluyendo los puntos de conexión de SQL Analytics y los almacenes de Fabric.
  • fabric.functions.FabricLakehouseClient: permite trabajar con almacenes de lago, con una manera de conectarse tanto a tablas de almacén de lago como a archivos de almacén de lago.

Para hacer referencia a una conexión a un origen de datos, es necesario usar el decorador @udf.connection. Aplíquelo en cualquiera de los siguientes formatos:

  • @udf.connection(alias="<alias for data connection>", argName="sqlDB")
  • @udf.connection("<alias for data connection>", "<argName>")
  • @udf.connection("<alias for data connection>")

Los argumentos de @udf.connection son:

  • argName, el nombre de la variable que usa la conexión en la función.
  • alias, el alias de la conexión que agregó con el menú Administrar conexiones.
  • Si argName y alias tuvieran el mismo valor, use @udf.connection("<alias and argName for the data connection>").

Ejemplo

# Where demosqldatabase is the argument name and the alias for my data connection used for this function
@udf.connection("demosqldatabase")
@udf.function()
def read_from_sql_db(demosqldatabase: fn.FabricSqlConnection)-> list:
  # Connect to the SQL database
  connection = demosqldatabase.connect()
  cursor = connection.cursor()
  
  # Replace with the query you want to run
  query = "SELECT * FROM (VALUES ('John Smith', 31), ('Kayla Jones', 33)) AS Employee(EmpName, DepID);"
  
  # Execute the query
  cursor.execute(query)
  
  # Fetch all results
  results = cursor.fetchall()
  
  # Close the cursor and connection
  cursor.close()
  connection.close()
  
  return results

Conexiones genéricas para elementos de Fabric o recursos de Azure

El SDK admite conexiones genéricas que permiten crear conexiones a elementos de Fabric o recursos de Azure mediante la identidad del propietario del elemento de User Data Functions. Esta función genera un token de identificación de Microsoft Entra con la identidad del propietario del elemento y un tipo de audiencia proporcionado. Este token se usa para autenticarse con elementos de Fabric o recursos de Azure que admiten ese tipo de audiencia. Este enfoque proporciona una experiencia de programación similar al uso de objetos de conexiones administradas desde la característica Administrar conexiones , pero solo para el tipo de público proporcionado en la conexión.

Esta característica usa el @udf.generic_connection() decorador con los parámetros siguientes:

Parámetro Descripción Importancia
argName Nombre de la variable que se pasa a la función. El usuario debe especificar esta variable en los argumentos de su función y usar el tipo de fn.FabricItem para ella. Por ejemplo, si es argName=CosmosDb, la función debe contener este argumento. cosmosDb: fn.FabricItem
audienceType Tipo de público para el que se crea la conexión. Este parámetro está asociado al tipo de elemento fabric o servicio de Azure y determina el cliente usado para la conexión. Los valores permitidos para este parámetro son CosmosDb o KeyVault.

Conexión al contenedor de Cosmos DB de Fabric mediante una conexión genérica

Las conexiones genéricas admiten elementos nativos de Fabric Cosmos DB utilizando el tipo de audiencia CosmosDB. El SDK de Funciones de Datos de Usuario incluido proporciona un método auxiliar denominado get_cosmos_client que obtiene un cliente singleton de Cosmos DB para cada invocación.

Puede conectarse a un elemento de Fabric Cosmos DB mediante una conexión genérica siguiendo estos pasos:

  1. Vaya al portal de Fabric, busque el área de trabajo y abra el elemento de funciones de datos de usuario. Seleccione Administración de bibliotecas, busque la azure-cosmos biblioteca e instálela. Para obtener más información, consulte Administración de bibliotecas.

  2. Vaya a la configuración del Fabric Cosmos DB elemento.

    Captura de pantalla que muestra la ubicación del botón configuración de Fabric Cosmos DB.

  3. Recupere la dirección URL del punto de conexión de Fabric Cosmos DB.

    Captura de pantalla que muestra la dirección URL del punto de conexión de Fabric Cosmos DB.

  4. Vaya al elemento Funciones de datos de usuario. Use el código de ejemplo siguiente para conectarse al contenedor de Fabric Cosmos DB y ejecutar una consulta de lectura mediante el conjunto de datos de ejemplo de Cosmos DB. Reemplace los valores de las variables siguientes:

    • COSMOS_DB_URI con tu punto de conexión de Fabric Cosmos DB.
    • DB_NAME con el nombre del elemento de Fabric Cosmos DB.
    from fabric.functions.cosmosdb import get_cosmos_client
import json

@udf.generic_connection(argName="cosmosDb", audienceType="CosmosDB")
@udf.function()
def get_product_by_category(cosmosDb: fn.FabricItem, category: str) -> list:

    COSMOS_DB_URI = "YOUR_COSMOS_DB_URL"
    DB_NAME = "YOUR_COSMOS_DB_NAME" # Note: This is the Fabric item name
    CONTAINER_NAME = "SampleData" # Note: This is your container name. In this example, we are using the SampleData container.

    cosmosClient = get_cosmos_client(cosmosDb, COSMOS_DB_URI)

    # Get the database and container
    database = cosmosClient.get_database_client(DB_NAME)
    container = database.get_container_client(CONTAINER_NAME)

    query = 'select * from c WHERE c.category=@category' #"select * from c where c.category=@category"
    parameters = [
        {
            "name": "@category", "value": category
        }
    ]
    results = container.query_items(query=query, parameters=parameters)
    items = [item for item in results]

    logging.info(f"Found {len(items)} products in {category}")

    return json.dumps(items)

  5. Pruebe o ejecute esta función proporcionando un nombre de categoría, como Accessory en los parámetros de invocación.

Nota:

También puede usar estos pasos para conectarse a una base de datos de Azure Cosmos DB mediante la dirección URL de la cuenta y los nombres de base de datos. La cuenta de propietario de User Data Functions necesitaría permisos de acceso a esa cuenta de Azure Cosmos DB.

Conexión a Azure Key Vault mediante una conexión genérica

Las conexiones genéricas admiten la conexión a un Azure Key Vault mediante el tipo de audiencia KeyVault. Este tipo de conexión requiere que el propietario de Fabric User Data Functions tenga permisos para conectarse a Azure Key Vault. Puede usar esta conexión para recuperar claves, secretos o certificados por nombre.

Puede conectarse a Azure Key Vault para recuperar un secreto de cliente para llamar a una API mediante una conexión genérica siguiendo estos pasos:

  1. Vaya al portal de Fabric, busque el área de trabajo y abra el elemento de funciones de datos de usuario. Seleccione Administración de bibliotecas y busque e instale las requests bibliotecas y azure-keyvault-secrets . Para obtener más información, consulte Administración de bibliotecas.

  2. Vaya a su recurso de Azure Key Vault en el portal de Azure y recupere el Vault URI y el nombre de su clave, secreto o certificado.

    Captura de pantalla que muestra la dirección URL y los valores del punto de conexión de Azure Key Vault.

  3. Vuelva al elemento Fabric User Data Functions y use este ejemplo. En este ejemplo, se recupera un secreto de Azure Key Vault para conectarse a una API pública. Reemplace el valor de las variables siguientes:

    • KEY_VAULT_URL con el Vault URI que recuperaste en el paso anterior.
    • KEY_VAULT_SECRET_NAME con el nombre de su secreto.
    • API_URL variable con la dirección URL de la API a la que desea conectarse. En este ejemplo se supone que se va a conectar a una API pública que acepta solicitudes GET y toma los siguientes parámetros api-key y request-body.
    from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
import requests

@udf.generic_connection(argName="keyVaultClient", audienceType="KeyVault")
@udf.function()
def retrieveNews(keyVaultClient: fn.FabricItem, requestBody:str) -> str:
    KEY_VAULT_URL = 'YOUR_KEY_VAULT_URL'
    KEY_VAULT_SECRET_NAME= 'YOUR_SECRET'
    API_URL = 'YOUR_API_URL'

    credential = keyVaultClient.get_access_token()

    client = SecretClient(vault_url=KEY_VAULT_URL, credential=credential)

    api_key = client.get_secret(KEY_VAULT_SECRET_NAME).value

    api_url = API_URL
    params = {
        "api-key": api_key,
        "request-body": requestBody
    }

    response = requests.get(api_url, params=params)

    data = "" 

    if response.status_code == 200:
        data = response.json()
    else:
        print(f"Error {response.status_code}: {response.text}")

    return f"Response: {data}"

  4. Pruebe o ejecute esta función proporcionando un cuerpo de solicitud en el código.

Características avanzadas

El modelo de programación define patrones avanzados que proporcionan un mayor control sobre las funciones. El SDK implementa estos patrones a través de clases y métodos que le permiten:

  • Obtener acceso a los metadatos de invocación de quién llamó a tu función y cómo
  • Control de escenarios de error personalizados con respuestas de error estructuradas
  • Integración con bibliotecas de variables de Fabric para la administración centralizada de la configuración

Nota:

User Data Functions tiene límites de servicio para el tamaño de solicitud, el tiempo de espera de ejecución y el tamaño de respuesta. Para más información sobre estos límites y cómo se aplican, consulte Detalles y limitaciones del servicio.

Obtención de propiedades de invocación mediante UserDataFunctionContext

El SDK incluye el UserDataFunctionContext objeto . Este objeto contiene los metadatos de invocación de función y se puede usar para crear una lógica de aplicación específica para distintos mecanismos de invocación (como la invocación del portal frente a la invocación de la API REST).

En la tabla siguiente se muestran las propiedades del objeto UserDataFunctionContext:

Nombre de propiedad Tipo de datos Descripción
invocation_id cuerda / cadena GUID único asociado a la invocación del elemento de funciones de datos de usuario.
usuario_ejecutor objeto Metadatos de la información del usuario utilizada para autorizar la invocación.

El objeto executing_user contiene la siguiente información:

Nombre de propiedad Tipo de datos Descripción
OID cadena (GUID) Id. de objeto del usuario, que es un identificador inmutable para el solicitante. Esta es la identidad comprobada del usuario o la entidad de servicio que se usa para invocar esta función en todas las aplicaciones.
Id. de inquilino cadena (GUID) Id. del inquilino en el que el usuario inició sesión.
NombreDeUsuarioPreferido cuerda / cadena El nombre de usuario preferido del usuario que invoca, tal y como lo establece el usuario. Este valor es mutable.

Para acceder al parámetro UserDataFunctionContext, use el siguiente decorador en la parte superior de la definición de función: @udf.context(argName="<parameter name>")

Ejemplo

@udf.context(argName="myContext")
@udf.function()
def getContext(myContext: fabric.functions.UserDataFunctionContext)-> str:
    logging.info('Python UDF trigger function processed a request.')
    return f"Hello oid = {myContext.executing_user['Oid']}, TenantId = {myContext.executing_user['TenantId']}, PreferredUsername = {myContext.executing_user['PreferredUsername']}, InvocationId = {myContext.invocation_id}"

Iniciar un error controlado con UserThrownError

Al desarrollar su función, puede generar una respuesta de error esperada mediante UserThrownError clase disponible en el SDK. Un uso de esta clase es administrar casos en los que las entradas proporcionadas por el usuario no superan las reglas de validación empresarial.

Ejemplo

import datetime

@udf.function()
def raise_userthrownerror(age: int)-> str:
    if age < 18:
        raise fn.UserThrownError("You must be 18 years or older to use this service.", {"age": age})

    return f"Welcome to Fabric Functions at {datetime.datetime.now()}!"

El UserThrownError constructor de clase toma dos parámetros:

  • Message: esta cadena se devuelve como mensaje de error a la aplicación que invoca esta función.
  • Se devuelve un diccionario de propiedades a la aplicación que invoca esta función.

Obtención de variables de las bibliotecas de Fabric

Una biblioteca de variables de Fabric en Microsoft Fabric es un repositorio centralizado para administrar variables que se pueden usar en distintos elementos dentro de un área de trabajo. Permite a los desarrolladores personalizar y compartir configuraciones de elementos de forma eficaz. Si aún no tiene una biblioteca de variables, consulte Creación y administración de bibliotecas de variables.

Para usar una biblioteca de variables en las funciones, agregue una conexión a ella desde el elemento de funciones de datos de usuario. Las bibliotecas de variables aparecen en el catálogo de OneLake junto con orígenes de datos como bases de datos SQL y lakehouses.

Siga estos pasos para usar bibliotecas de variables en las funciones:

  1. En el elemento de funciones de datos de usuario, agregue una conexión a la biblioteca de variables. En el catálogo de OneLake, busque y seleccione la biblioteca de variables y, a continuación, seleccione Conectar. Tenga en cuenta el alias que genera Fabric para la conexión.
  2. Agregue un decorador de conexión para el elemento de biblioteca de variables. Por ejemplo, @udf.connection(argName="varLib", alias="<My Variable Library Alias>") y reemplace el alias por la conexión recién agregada para el elemento de la biblioteca de variables.
  3. En la definición de función, incluya un argumento con el tipo fn.FabricVariablesClient. Este cliente proporciona métodos que necesita para trabajar con el elemento de biblioteca de variables.
  4. Utilice el método getVariables() para obtener todas las variables de la biblioteca de variables.
  5. Para leer los valores de las variables, use ["variable-name"] o .get("variable-name").

Ejemplo

En este ejemplo simulamos un escenario de configuración para una producción y un entorno de desarrollo. Esta función establece una ruta de acceso de almacenamiento en función del entorno seleccionado mediante un valor recuperado de la biblioteca de variables. La biblioteca de variables contiene una variable denominada ENV donde los usuarios pueden establecer un valor de dev o prod.

@udf.connection(argName="varLib", alias="<My Variable Library Alias>")
@udf.function()
def get_storage_path(dataset: str, varLib: fn.FabricVariablesClient) -> str:
    """
    Description: Determine storage path for a dataset based on environment configuration from Variable Library.
    
    Args:
        dataset_name (str): Name of the dataset to store.
        varLib (fn.FabricVariablesClient): Fabric Variable Library connection.
    
    Returns:
        str: Full storage path for the dataset.
    """
    # Retrieve variables from Variable Library
    variables = varLib.getVariables()
    
    # Get environment and base paths
    env = variables.get("ENV")    
    dev_path = variables.get("DEV_FILE_PATH")
    prod_path = variables.get("PROD_FILE_PATH")
    
    # Apply environment-specific logic
    if env.lower() == "dev":
        return f"{dev_path}{dataset}/"
    elif env.lower() == "prod":
        return f"{prod_path}{dataset}/"
    else:
        return f"incorrect settings define for ENV variable"

Contenido relacionado

  • Last updated on