Zdieľať cez


Prehľad programovacieho modelu používateľských dátových funkcií štruktúry

Model programovania dátových funkcií používateľov Fabric definuje vzory a koncepty pre tvorbu funkcií vo Fabric.

SDK implementuje fabric-user-data-functions tento programovací model a poskytuje potrebnú funkcionalitu na tvorbu a publikovanie spustiteľných funkcií. Súprava SDK vám tiež umožňuje bezproblémovú integráciu s inými položkami v ekosystéme služby Fabric, ako sú napríklad zdroje údajov služby Fabric. Táto knižnica je verejne dostupná v PyPI a je predinštalovaná vo vašich používateľských dátových funkciách.

Tento článok vysvetľuje, ako použiť SDK na tvorbu funkcií, ktoré je možné volať z Fabric portálu, iných položiek Fabric alebo externých aplikácií pomocou REST API. Naučíte sa programovací model a kľúčové koncepty pomocou praktických príkladov.

Prepitné

Pre úplné podrobnosti o všetkých triedach, metódach a parametroch si pozrite referenčnú dokumentáciu SDK.

Getting started so SDK

Táto sekcia predstavuje základné komponenty User Data Functions SDK a vysvetľuje, ako štruktúrovať vaše funkcie. Naučíte sa o potrebných importoch, dekorátoroch a typoch vstupných a výstupných dát, ktoré vaše funkcie dokážu spracovať.

Súprava SDK funkcií údajov používateľov

fabric-user-data-functions SDK poskytuje základné komponenty, ktoré potrebujete na vytvorenie používateľských dátových funkcií v Pythone.

Požadované importy a inicializácia

Každý súbor používateľských dátových funkcií musí importovať fabric.functions modul a inicializovať kontext vykonávania:

import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()

Dekoratér @udf.function()

Funkcie označené dekorátorom @udf.function() je možné vyvolať z portálu Fabric, iného predmetu Fabric alebo externej aplikácie. Funkcie s týmto dekorátorom musia špecifikovať typ návratu.

Príklad:

@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()}!"

Pomocné funkcie

Python metódy bez dekorátora @udf.function() sa nedajú volať priamo. Môžu byť privolané len z ozdobených funkcií a slúžia ako pomocné funkcie.

Príklad:

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

Podporované typy vstupov

Môžete definovať vstupné parametre pre funkciu, ako sú napríklad primitívne typy údajov, ako napríklad str, int, float atď. Podporované typy vstupných údajov sú:

typu JSON typ údajov jazyka Python
Povrázok Str
reťazca datetime Datetime
Boolean Bool
Numbers int, plavák
Pole list[], príklad zoznamu[int]
objektu Dict
objektu Dátový rámec pandas
Objekt alebo pole objektov séria pandas

Poznámka

Ak chcete použiť pandas DataFrame a Series typy, choďte do portálu Fabric, nájdite svoj pracovný priestor a otvorte položku používateľských dátových funkcií. Vyberte správu knižnice, vyhľadajte fabric-user-data-functions balík a aktualizujte ho na verziu 1.0.0 alebo novšiu.

Príklad textu požiadavky pre podporované typy vstupov:

{
  "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}
  ]
}

Podporované typy výstupu

Podporované typy údajov výstupu sú:

typ údajov jazyka Python
Str
Datetime
Bool
int, plavák
list[data-type], napríklad list[int]
Dict
žiadne
séria pandas
Dátový rámec pandas

Funkcie písania

Požiadavky na syntax a obmedzenia

Pri písaní funkcií používateľských dát musíte dodržiavať špecifické syntaktické pravidlá, aby vaše funkcie fungovali správne.

Pomenovanie parametrov

  • Použite camelCase: Názvy parametrov musia používať konvenciu pomenovania camelCase a nesmú obsahovať podčiarknutia. Napríklad použite productName namiesto .product_name
  • Rezervované kľúčové slová: Nemôžete použiť rezervované Python kľúčové slová ani nasledujúce Fabric-špecifické kľúčové slová ako názvy parametrov alebo funkcií: req, context, a reqInvocationId.

Požiadavky na parametre

  • Požadované typové anotácie: Všetky parametre musia obsahovať typové anotácie (napríklad name: str).
  • Predvolené hodnoty: Podporované sú predvolené hodnoty parametrov. V Fabric používateľských dátových funkciách môžete definovať predvolené argumenty, aby bolo jednoduchšie volať a udržiavať váš kód. Predvolené hodnoty podporujú bežné JSON-serializovateľné typy, vrátane reťazcov, booleanov, čísel (int, float), polí (zoznamov) a objektov (slovníkov). Syntax
        @udf.function()
        def function_name(param1: type = value1, param2: type = value2, listparam: list | None = None, ...) -> output_type:
            # function body
    
    Upozorňujeme, že predvolená hodnota musí byť JSON serializovateľná. Napríklad vnorené zoznamy ako [1, 2, [3]] sú povolené, zatiaľ čo vnorené množiny alebo n-tice nie sú podporované. Pre zoznamové alebo slovníkové predvolené nastavenia uprednostňujte použitie None v podpise a priradenie skutočného predvoleného vo funkcii.

Požiadavky na funkciu

  • Požadovaný typ návratu: Funkcie s dekorátorom @udf.function() musia špecifikovať anotáciu typu návratu (napríklad -> str).
  • Povinné importy: Príkaz import fabric.functions as fn a inicializácia sú potrebné, udf = fn.UserDataFunctions() aby vaše funkcie fungovali.

Príklad správnej syntaxe

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

Ako napísať asynchrónne funkcie

Pridajte asynchrónny dekorátor s definíciou funkcie v kóde. Vďaka funkcii async môžete zlepšiť odozvu a efektivitu aplikácie spracovávaním viacerých úloh naraz. Sú ideálne na správu veľkého objemu vstupno-výstupných operácií. Funkcia v tomto príklade prečíta súbor CSV z jazera pomocou pandy. Funkcia použije názov súboru ako vstupný parameter.

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}"

Práca s dátami

Údajové pripojenia k zdrojom údajov služby Fabric

SDK vám umožňuje odkazovať na dátové spojenia bez potreby zapisovať spojovacie reťazce do kódu. Knižnica fabric.functions poskytuje dva spôsoby na spracovanie údajových pripojení:

  • fabric.functions.FabricSqlConnection: Umožňuje pracovať s databázami SQL v službe Fabric vrátane koncových bodov služby SQL Analytics a skladov služby Fabric.
  • fabric.functions.FabricLakehouseClient: Umožňuje vám pracovať s lakehouses, so spôsobom, ako sa pripojiť k obom lakehouse tabuliek a Lakehouse súbory.

Ak chcete odkazovať na pripojenie k zdroju údajov, musíte použiť dekorátor @udf.connection. Môžete ju použiť v ktoromkoľvek z nasledujúcich formátov:

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

Argumenty pre @udf.connection sú:

  • argName, názov premennej, ktorá pripojenie používa vo vašej funkcii.
  • alias, alias pripojenia, ktoré ste pridali, pomocou ponuky Spravovať pripojenia.
  • Ak argName a alias majú rovnakú hodnotu, môžete použiť @udf.connection("<alias and argName for the data connection>").

Príklad

# 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

Generické prepojenia pre Fabric položky alebo Azure zdroje

SDK podporuje generické spojenia, ktoré umožňujú vytvárať spojenia k Fabric položkám alebo Azure zdrojom pomocou identity vlastníka položky v User Data Functions. Táto funkcia generuje Microsoft Entra ID token s identitou vlastníka položky a zadaným typom publika. Tento token sa používa na autentifikáciu pomocou Fabric položiek alebo Azure zdrojov, ktoré podporujú daný typ publika. Tento prístup poskytuje podobný programátorský zážitok ako používanie spravovaných prepojení objektov z funkcie Spravovať spojenia , ale len pre daný typ publika v spojení.

Táto funkcia @udf.generic_connection() používa dekoratér s nasledujúcimi parametrami:

Parameter Popis Value
argName Názov premennej, ktorá sa odovzdáva funkcii. Používateľ musí špecifikovať túto premennú v argumentoch svojej funkcie a použiť pre ňu typ fn.FabricItem Ak by napríklad funkcia argName=CosmosDbmala obsahovať tento argument cosmosDb: fn.FabricItem
audienceType Typ cieľovej skupiny, pre ktorú je pripojenie vytvorené. Tento parameter je priradený k typu Fabric položky alebo Azure služby a určuje klienta použitého pre spojenie. Povolené hodnoty pre tento parameter sú CosmosDb alebo KeyVault.

Pripojenie ku kontajneru služby Fabric Cosmos DB pomocou všeobecného pripojenia

Všeobecné pripojenia podporujú natívne položky služby Fabric Cosmos DB pomocou CosmosDB typu cieľovej skupiny. Zahrnutá súprava User Data Functions SDK poskytuje pomocnú metódu s názvom get_cosmos_client Helper, ktorá načíta jedného klienta Cosmos DB pre každé vyvolanie.

K položke služby Fabric Cosmos DB sa môžete pripojiť pomocou všeobecného pripojenia podľa týchto krokov:

  1. Choďte do portálu Fabric, nájdite svoj pracovný priestor a otvorte položku funkcie používateľských dát. Vyberte Library management, vyhľadajte knižnicu azure-cosmos a nainštalujte ju. Pre viac informácií pozri Spravovať knižnice.

  2. Prejdite do nastavení položiek služby Fabric Cosmos DB .

    Snímka obrazovky zobrazujúca umiestnenie tlačidla nastavení služby Fabric Cosmos DB.

  3. Načítajte URL adresu koncového bodu služby Fabric Cosmos DB.

    Snímka obrazovky zobrazujúca URL adresu koncového bodu služby Fabric Cosmos DB.

  4. Prejdite na položku Funkcie používateľských údajov. Pomocou nasledujúceho vzorového kódu sa pripojte ku kontajneru služby Fabric Cosmos DB a spustite dotaz na čítanie pomocou vzorovej množiny údajov služby Cosmos DB. Nahraďte hodnoty nasledujúcich premenných:

    • COSMOS_DB_URI s koncovým bodom služby Fabric Cosmos DB.
    • DB_NAME s názvom položky služby 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. Otestujte alebo spustite túto funkciu zadaním názvu kategórie, napríklad Accessory v parametroch vyvolania.

Poznámka

Tieto kroky môžete použiť aj na pripojenie k databáze Azure Cosmos DB pomocou URL adresy účtu a názvov databáz. Účet vlastníka User Data Functions by potreboval access oprávnenia k tomuto účtu Azure Cosmos DB.

Connect to Azure Key Vault pomocou generického pripojenia

Generické spojenia podporujú pripojenie k Azure Key Vault pomocou typu KeyVault publika. Tento typ pripojenia vyžaduje, aby vlastník Fabric User Data Functions mal oprávnenia na pripojenie k Azure Key Vault. Toto pripojenie môžete použiť na načítanie kľúčov, tajných kódov alebo certifikátov podľa názvu.

Môžete sa pripojiť k Azure Key Vault a získať klientské tajomstvo na volanie API pomocou generického pripojenia podľa týchto krokov:

  1. Choďte do portálu Fabric, nájdite svoj pracovný priestor a otvorte položku funkcie používateľských dát. Vyberte Library management, potom vyhľadajte a nainštalujte knižnice requests a azure-keyvault-secrets. Pre viac informácií pozri Spravovať knižnice.

  2. Choďte do svojho Azure Key Vault zdroja v Azure portal a získajte Vault URI a názov vášho kľúča, tajomstva alebo certifikátu.

    Snímka obrazovky zobrazujúca URL a hodnoty Azure Key Vault endpointu.

  3. Go back na položku Fabric User Data Functions a použite túto ukážku. V tejto ukážke získavame tajomstvo z Azure Key Vault, aby sme sa pripojili k verejnému API. Nahraďte hodnotu nasledujúcich premenných:

    • KEY_VAULT_URL s možnosťou, ktorú Vault URI ste načítali v predchádzajúcom kroku.
    • KEY_VAULT_SECRET_NAME s názvom vášho tajomstva.
    • API_URL s URL adresou rozhrania API, ku ktorému sa chcete pripojiť. V tejto ukážke sa predpokladá, že sa pripájate k verejnému rozhraniu API, ktoré prijíma požiadavky GET a preberá nasledujúce parametre api-key a 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. Otestujte alebo spustite túto funkciu zadaním tela požiadavky v kóde.

Rozšírené funkcie

Programovací model definuje pokročilé vzory, ktoré vám dávajú väčšiu kontrolu nad vašimi funkciami. SDK implementuje tieto vzory prostredníctvom tried a metód, ktoré vám umožňujú:

  • Access metadáta volania o tom, kto a ako volal vašu funkciu
  • Riešenie vlastných chybových scenárov pomocou štruktúrovaných odpovedí na chyby
  • Integrácia s knižnicami premenných Fabric pre centralizovanú správu konfigurácie

Poznámka

User Data Functions má limity služieb pre veľkosť požiadavky, časový limit vykonávania a veľkosť odpovede. Podrobnosti o týchto obmedzeniach a spôsobe ich uplatňovania nájdete v sekcii Podrobnosti o službách a obmedzeniach.

Získanie vlastností vyvolania pomocou UserDataFunctionContext

SDK obsahuje UserDataFunctionContext objekt. Tento objekt obsahuje metadáta volania funkcie a môže byť použitý na vytvorenie špecifickej logiky aplikácie pre rôzne mechanizmy volania (napríklad volanie portálu verzus REST API).

Nasledujúca tabuľka zobrazuje vlastnosti pre UserDataFunctionContext objekt:

Názov vlastnosti Typ údajov Popis
invocation_id reťazec Jedinečný identifikátor GUID zviazaný s vyvolanie položky údajových funkcií používateľa.
executing_user objekt Metaúdaje informácií o používateľovi použité na overenie vyvolania.

Objekt executing_user obsahuje nasledujúce informácie:

Názov vlastnosti Typ údajov Popis
Oid Reťazec (GUID) ID objektu používateľa, ktoré je nemeniteľným identifikátorom žiadateľa. Ide o overenú identitu používateľa alebo objektu služby, ktorá sa použila na vyvolanie tejto funkcie v rámci aplikácií.
ID nájomníka Reťazec (GUID) ID nájomníka, do ktorému je používateľ prihlásený.
PreferredUsername reťazec Preferované meno používateľa na vyvolanie, ako je nastavené používateľom. Táto hodnota je mutable.

Na access parametra UserDataFunctionContext musíte použiť nasledujúci dekorátor v hornej časti definície funkcie: @udf.context(argName="<parameter name>")

Príklad

@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}"

Hodiť spracovanú chybu s UserThrownError

Pri vývoji funkcie môžete vyvolať očakávanú chybovú odpoveď pomocou triedy dostupnej UserThrownError v SDK. Jedným z využití tejto triedy je riadenie prípadov, keď vstupy poskytnuté používateľom neprejdú pravidlami overovania podniku.

Príklad

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()}!"

Konštruktor UserThrownError triedy má dva parametre:

  • Message: Tento reťazec sa vráti ako chybové hlásenie aplikácii, ktorá vyvoláva túto funkciu.
  • Slovník vlastností sa vráti do aplikácie, ktorá vyvoláva túto funkciu.

Získanie premenných z knižníc premenných štruktúry

Knižnica premenných Fabric variable library in Microsoft Fabric je centralizovaný repozitár na správu premenných, ktoré môžu byť použité naprieč rôznymi položkami v pracovnom priestore. Umožňuje vývojárom efektívne prispôsobovať a zdieľať konfigurácie položiek. Ak ešte nemáte knižnicu premenných, pozrite si Vytvoriť a spravovať knižnice premenných.

Ak chcete použiť knižnicu premenných vo svojich funkciách, pridáte k nej spojenie z položky používateľských dátových funkcií. Knižnice premenných sa objavujú v katalógu OneLake spolu s dátovými zdrojmi, ako sú SQL databázy a jazerné domy.

Postupujte podľa týchto krokov pri používaní knižníc premenných vo vašich funkciách:

  1. V položke funkcie používateľských dát pridajte spojenie do knižnice premenných. V katalógu OneLake nájdite a vyberte svoju knižnicu premenných, potom vyberte Pripojiť. Všimnite si alias , ktorý Fabric generuje pre spojenie.
  2. Pridanie dekorátora spojenia pre položku knižnice premenných. Napríklad @udf.connection(argName="varLib", alias="<My Variable Library Alias>") a nahradiť alias novo pridaným pripojením pre položku knižnice premenných.
  3. Do definície funkcie zahrňte argument s typom fn.FabricVariablesClient. Tento klient poskytuje metódy, ktoré potrebujete na prácu s položkou knižnice premenných.
  4. Použite getVariables() metódu na získanie všetkých premenných z knižnice premenných.
  5. Ak chcete prečítať hodnoty premenných, použite buď ["variable-name"] alebo .get("variable-name").

Príklad

V tomto príklade simulujeme konfiguračný scenár pre produkčné a vývojové prostredie. Táto funkcia nastavuje storage path v závislosti od vybraného prostredia pomocou hodnoty získanej z Variable Library. Knižnica premenných obsahuje premennú s názvom ENV , kde môžu používatelia nastaviť hodnotu dev alebo 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"