Nuta
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować się zalogować lub zmienić katalog.
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować zmienić katalogi.
Model programowania funkcji danych użytkownika sieci szkieletowej definiuje wzorce i pojęcia dotyczące tworzenia funkcji w sieci szkieletowej.
Zestaw fabric-user-data-functions SDK implementuje ten model programowania, zapewniając niezbędne funkcje do tworzenia i publikowania funkcji możliwych do uruchomienia. Zestaw SDK umożliwia również bezproblemową integrację z innymi elementami w ekosystemie sieci Szkieletowej, takimi jak źródła danych sieci szkieletowej. Ta biblioteka jest publicznie dostępna w PyPI i jest wstępnie zainstalowana w elementach funkcji danych użytkownika.
W tym artykule wyjaśniono, jak używać zestawu SDK do tworzenia funkcji, które mogą być wywoływane z poziomu portalu sieci szkieletowej, innych elementów sieci szkieletowej lub aplikacji zewnętrznych przy użyciu interfejsu API REST. Poznasz model programowania i kluczowe pojęcia z praktycznymi przykładami.
Wskazówka
Aby uzyskać szczegółowe informacje na temat wszystkich klas, metod i parametrów, zobacz dokumentację referencyjną zestawu SDK.
Wprowadzenie do zestawu SDK
W tej sekcji przedstawiono podstawowe składniki zestawu SDK (Software Development Kit) funkcji danych użytkownika i wyjaśniono, jak strukturyzować swoje funkcje. Dowiesz się o wymaganych importach, dekoratorach oraz typach danych wejściowych i wyjściowych, które mogą obsłużyć funkcje.
Zestaw SDK do funkcji danych użytkownika
Zestaw fabric-user-data-functions SDK udostępnia podstawowe składniki potrzebne do tworzenia funkcji danych użytkownika w języku Python.
Wymagane importy i inicjowanie
Każdy plik funkcji danych użytkownika musi zaimportować fabric.functions moduł i zainicjować kontekst wykonywania:
import datetime
import fabric.functions as fn
import logging
udf = fn.UserDataFunctions()
Dekorator @udf.function()
Funkcje oznaczone dekoratorem @udf.function() mogą być wywoływane z portalu Fabric, innego elementu Fabric lub aplikacji zewnętrznej. Funkcje z tym dekoratorem muszą określać typ zwracany.
Przykład:
@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()}!"
Funkcje pomocnicze
Metody języka Python bez @udf.function() dekoratora nie mogą być wywoływane bezpośrednio. Mogą być wywoływane tylko z funkcji dekorowanych i służyć jako funkcje pomocnicze.
Przykład:
def uppercase_name(name: str) -> str:
return name.upper()
Obsługiwane typy danych wejściowych
Można zdefiniować parametry wejściowe dla funkcji, takie jak typy danych pierwotnych, takie jak str, int, float itp. Obsługiwane typy danych wejściowych to:
| Typ JSON | typ danych języka Python |
|---|---|
| String | Str |
| Ciąg daty/godziny | data/godzina |
| Boolean | Bool |
| Liczby | int, float |
| Tablica | list[], przykładowa lista[int] |
| Obiekt | słownik |
| Obiekt | Ramka danych biblioteki pandas |
| Obiekt lub tablica obiektów | Pandas Series |
Uwaga / Notatka
Aby użyć typów DataFrame i Series pandas, przejdź do portalu Fabric, znajdź swój obszar roboczy i otwórz element funkcji danych użytkownika. Wybierz pozycję Zarządzanie biblioteką, wyszukaj fabric-user-data-functions pakiet i zaktualizuj go do wersji 1.0.0 lub nowszej.
Przykład treści żądania dla obsługiwanych typów danych wejściowych:
{
"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}
]
}
Obsługiwane typy danych wyjściowych
Obsługiwane typy danych wyjściowych to:
| typ danych języka Python |
|---|
| Str |
| data/godzina |
| Bool |
| int, float |
| list[data-type], na przykład list[int] |
| słownik |
| Żaden |
| Pandas Series |
| Ramka danych biblioteki pandas |
Pisanie funkcji
Wymagania i ograniczenia składni
Podczas pisania funkcji danych użytkownika należy postępować zgodnie z określonymi regułami składni, aby upewnić się, że funkcje działają prawidłowo.
Nazewnictwo parametrów
-
Użyj camelCase: Nazwy parametrów muszą używać konwencji nazewnictwa camelCase i nie mogą zawierać podkreśleń. Na przykład użyj
productNamezamiastproduct_name. -
Zastrzeżone słowa kluczowe: nie można używać zarezerwowanych słów kluczowych języka Python ani następujących słów kluczowych specyficznych dla sieci szkieletowej jako nazw parametrów lub nazw funkcji:
req,contextireqInvocationId.
Wymagania dotyczące parametrów
-
Brak wartości domyślnych: domyślne wartości parametrów nie są obsługiwane. Wszystkie parametry są wymagane podczas wywoływania funkcji. Na przykład następująca funkcja zgłasza błąd składni:
# 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}." -
Wymagane adnotacje typu: Wszystkie parametry muszą zawierać adnotacje typu (na przykład
name: str).
Wymagania dotyczące funkcji
-
Wymagany typ zwracany: funkcje z
@udf.function()dekoratorem muszą określać adnotację typu zwracanego (na przykład-> str). -
Wymagane importy: instrukcja
import fabric.functions as fniudf = fn.UserDataFunctions()inicjalizacja są wymagane, aby funkcje działały.
Przykład poprawnej składni
@udf.function()
def process_order(orderNumber: int, customerName: str, orderDate: str) -> dict:
return {
"order_id": orderNumber,
"customer": customerName,
"date": orderDate,
"status": "processed"
}
Jak napisać funkcję asynchroniową
Dodaj dekorator asynchroniczny z definicją funkcji w kodzie.
async Dzięki funkcji można poprawić czas odpowiedzi i wydajność aplikacji, obsługując jednocześnie wiele zadań. Doskonale nadają się do zarządzania dużymi ilościami operacji związanych z operacjami we/wy. Ta przykładowa funkcja odczytuje plik CSV z magazynu lakehouse przy użyciu biblioteki pandas. Funkcja przyjmuje nazwę pliku jako parametr wejściowy.
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}"
Praca z danymi
Połączenia danych ze źródłami danych Fabric
Zestaw SDK umożliwia odwołanie się do połączeń danych bez konieczności pisania parametrów połączenia w kodzie. Biblioteka fabric.functions udostępnia dwa sposoby na obsługę połączeń danych:
- fabric.functions.FabricSqlConnection: Umożliwia pracę z bazami danych SQL w usłudze Fabric, w tym punktami końcowymi usługi SQL Analytics i repozytoriami Fabric.
- fabric.functions.FabricLakehouseClient: Umożliwia pracę z Lakehouses, oferując możliwość połączenia z tabelami i plikami Lakehouse.
Aby odwołać się do połączenia ze źródłem danych, należy użyć dekoratora @udf.connection. Można go zastosować w dowolnym z następujących formatów:
@udf.connection(alias="<alias for data connection>", argName="sqlDB")@udf.connection("<alias for data connection>", "<argName>")@udf.connection("<alias for data connection>")
Argumenty dla @udf.connection są następujące:
-
argName, nazwa zmiennej używanej przez połączenie w funkcji. -
alias, alias połączenia dodanego za pomocą menu Zarządzanie połączeniami . - Jeśli
argNameialiasmają tę samą wartość, możesz użyć@udf.connection("<alias and argName for the data connection>").
Przykład
# 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
Ogólne połączenia dla elementów Fabric lub zasobów Azure
Zestaw SDK obsługuje połączenia ogólne, które umożliwiają tworzenie połączeń z elementami sieci szkieletowej lub zasobami platformy Azure przy użyciu tożsamości właściciela elementu usługi User Data Functions. Ta funkcja generuje token identyfikatora Entra firmy Microsoft z tożsamością właściciela elementu i podanym typem odbiorców. Ten token służy do uwierzytelniania za pomocą elementów sieci szkieletowej lub zasobów platformy Azure, które obsługują ten typ odbiorców. Takie podejście zapewnia podobne środowisko programowania do używania obiektów połączeń zarządzanych z funkcji Zarządzanie połączeniami , ale tylko dla podanego typu odbiorców w połączeniu.
Ta funkcja używa dekoratora @udf.generic_connection() z następującymi parametrami:
| Parameter | Opis | Wartość |
|---|---|---|
argName |
Nazwa zmiennej przekazanej do funkcji. Użytkownik musi określić tę zmienną w argumentach funkcji i użyć typu fn.FabricItem dla niej |
Na przykład jeśli element argName=CosmosDb, funkcja powinna zawierać ten argument cosmosDb: fn.FabricItem |
audienceType |
Typ odbiorców, dla których jest tworzone połączenie. Ten parametr jest skojarzony z typem elementu sieci szkieletowej lub usługi platformy Azure i określa klienta używanego do połączenia. | Dozwolone wartości tego parametru to CosmosDb lub KeyVault. |
Podłącz się do kontenera Cosmos DB w Fabric przy użyciu połączenia ogólnego
Połączenia ogólne obsługują natywne elementy usługi Cosmos DB Fabric, korzystając z typu odbiorców CosmosDB. Dołączony zestaw SDK funkcji danych użytkownika udostępnia metodę pomocnika o nazwie get_cosmos_client, która uzyskuje pojedynczego klienta usługi Cosmos DB dla każdego wywołania.
Aby nawiązać połączenie z elementem usługi Fabric Cosmos DB , możesz nawiązać połączenie ogólne, wykonując następujące kroki:
Przejdź do portalu Fabric, znajdź swój obszar roboczy i otwórz element funkcji danych użytkownika. Wybierz pozycję Zarządzanie biblioteką, wyszukaj bibliotekę
azure-cosmosi zainstaluj ją. Aby uzyskać więcej informacji, zobacz Zarządzanie bibliotekami.Przejdź do ustawień elementu Cosmos DB w usłudze Fabric.
Pobierz adres URL punktu końcowego usługi Fabric dla Cosmos DB.
Przejdź do elementu Funkcje danych użytkownika. Użyj poniższego przykładowego kodu, aby nawiązać połączenie z kontenerem usługi Cosmos DB usługi Fabric i uruchomić zapytanie odczytu przy użyciu przykładowego zestawu danych usługi Cosmos DB. Zastąp wartości następujących zmiennych:
-
COSMOS_DB_URIz punktem końcowym usługi Fabric Cosmos DB. -
DB_NAMEz nazwą elementu 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)-
Przetestuj lub uruchom tę funkcję , podając nazwę kategorii, na przykład
Accessoryw parametrach wywołania.
Uwaga / Notatka
Możesz również użyć tych kroków, aby nawiązać połączenie z bazą danych usługi Azure Cosmos DB przy użyciu adresu URL konta i nazw baz danych. Konto właściciela usługi User Data Functions będzie potrzebować uprawnień dostępu do tego konta usługi Azure Cosmos DB.
Nawiązywanie połączenia z usługą Azure Key Vault przy użyciu połączenia ogólnego
Połączenia ogólne obsługują nawiązywanie połączenia z usługą Azure Key Vault przy użyciu KeyVault typu odbiorców. Ten typ połączenia wymaga, aby właściciel funkcji danych użytkownika Fabric miał uprawnienia do nawiązywania połączenia z Azure Key Vault. Możesz użyć tego połączenia, aby pobierać klucze, sekrety lub certyfikaty na podstawie nazwy.
Aby pobrać klucz tajny klienta w celu wywołania interfejsu API przy użyciu połączenia ogólnego, możesz nawiązać połączenie z usługą Azure Key Vault , wykonując następujące kroki:
Przejdź do portalu Fabric, znajdź swój obszar roboczy i otwórz element funkcji danych użytkownika. Wybierz Zarządzanie biblioteką, a następnie wyszukaj i zainstaluj biblioteki
requestsiazure-keyvault-secrets. Aby uzyskać więcej informacji, zobacz Zarządzanie bibliotekami.Przejdź do zasobu usługi Azure Key Vault w witrynie Azure portal i pobierz
Vault URInazwę swojego klucza, elementu tajnego lub certyfikatu.Wróć do Funkcji danych użytkownika Fabric i użyj tego przykładu. W tym przykładzie pobieramy tajną wartość z usługi Azure Key Vault, aby połączyć się z publicznym interfejsem API. Zastąp wartość następujących zmiennych:
-
KEY_VAULT_URLprzy użyciu elementuVault URI, który pobrano w poprzednim kroku. -
KEY_VAULT_SECRET_NAMEz nazwą twojego sekretu. -
API_URLzmienna z adresem URL interfejsu API, z którym chcesz nawiązać połączenie. W tym przykładzie założono, że nawiązujesz połączenie z publicznym interfejsem API, który akceptuje żądania GET i przyjmuje następujące parametryapi-keyirequest-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}"-
Przetestuj lub uruchom tę funkcję , podając treść żądania w kodzie.
Zaawansowane funkcje
Model programowania definiuje zaawansowane wzorce, które zapewniają większą kontrolę nad funkcjami. Zestaw SDK implementuje te wzorce za pomocą klas i metod, które umożliwiają:
- Uzyskaj dostęp do metadanych wywołania dotyczących tego, kto wywołał funkcję i w jaki sposób.
- Obsługa niestandardowych scenariuszy błędów z ustrukturyzowanymi odpowiedziami na błędy
- Integracja z bibliotekami zmiennych Fabric na potrzeby scentralizowanego zarządzania konfiguracją
Uwaga / Notatka
Funkcje danych użytkownika mają limity usługi dotyczące rozmiaru żądania, limitu czasu wykonywania i rozmiaru odpowiedzi. Aby uzyskać szczegółowe informacje na temat tych limitów i sposobu ich wymuszania, zobacz Szczegóły usługi i ograniczenia.
Uzyskiwanie właściwości wywołania przy użyciu polecenia UserDataFunctionContext
Zestaw SDK zawiera UserDataFunctionContext obiekt . Ten obiekt zawiera metadane wywołania funkcji i może służyć do tworzenia określonej logiki aplikacji dla różnych mechanizmów wywołania (takich jak wywołanie portalu a wywołanie interfejsu API REST).
W poniższej tabeli przedstawiono właściwości obiektu UserDataFunctionContext.
| Nazwa właściwości | Typ danych | Opis |
|---|---|---|
| invocation_id | ciąg | Unikatowy identyfikator GUID powiązany z wywołaniem elementu funkcji danych użytkownika. |
| użytkownik_wykonujący | obiekt | Metadane informacji użytkownika używanych do autoryzowania wywołania. |
Obiekt executing_user zawiera następujące informacje:
| Nazwa właściwości | Typ danych | Opis |
|---|---|---|
| Oid | string (GUID) | Identyfikator obiektu użytkownika, który jest niezmiennym identyfikatorem osoby żądającej. Jest to zweryfikowana tożsamość użytkownika lub jednostki usługi używanej do wywoływania tej funkcji w aplikacjach. |
| Identyfikator najemcy | string (GUID) | Identyfikator dzierżawcy, z którym użytkownik jest zalogowany. |
| Preferowana nazwa użytkownika | ciąg | Preferowana nazwa użytkownika wywołującego użytkownika ustawiona przez użytkownika. Ta wartość jest modyfikowalna. |
Aby uzyskać dostęp do parametru UserDataFunctionContext , należy użyć następującego dekoratora w górnej części definicji funkcji: @udf.context(argName="<parameter name>")
Przykład
@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}"
Zgłaszanie obsłużonego błędu za pomocą polecenia UserThrownError
Podczas tworzenia funkcji można zgłosić oczekiwaną odpowiedź o błędzie przy użyciu klasy dostępnej UserThrownError w zestawie SDK. Jednym z zastosowań tej klasy jest zarządzanie przypadkami, w których dane wejściowe dostarczone przez użytkownika nie spełniają reguł weryfikacji biznesowej.
Przykład
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()}!"
Konstruktor UserThrownError klasy przyjmuje dwa parametry:
-
Message: ten ciąg jest zwracany jako komunikat o błędzie do aplikacji wywołującej tę funkcję. - Słownik właściwości jest zwracany do aplikacji wywołującej tę funkcję.
Pobieranie zmiennych z bibliotek zmiennych Fabric
Biblioteka zmiennych Fabric w usłudze Microsoft Fabric to scentralizowane repozytorium do zarządzania zmiennymi, które mogą być używane w różnych elementach przestrzeni roboczej. Umożliwia deweloperom wydajne dostosowywanie i udostępnianie konfiguracji elementów. Jeśli nie masz jeszcze biblioteki zmiennych, zobacz Tworzenie bibliotek zmiennych i zarządzanie nimi.
Aby użyć biblioteki zmiennych w funkcjach, należy dodać do niej połączenie z elementu funkcji danych użytkownika. Biblioteki zmiennych są wyświetlane w katalogu OneLake wraz ze źródłami danych, takimi jak bazy danych SQL i lakehouse'y.
Wykonaj następujące kroki, aby użyć bibliotek zmiennych w funkcjach:
- W elemencie funkcji danych użytkownika dodaj połączenie do swojej biblioteki zmiennych. W katalogu OneLake znajdź i wybierz bibliotekę zmiennych, a następnie wybierz pozycję Połącz. Zwróć uwagę na Alias generowany przez Fabric dla połączenia.
- Dodaj dekorator połączeń dla elementu biblioteki zmiennych. Na przykład
@udf.connection(argName="varLib", alias="<My Variable Library Alias>")i zamień alias, wskazując na nowo dodane połączenie dla elementu biblioteki zmiennych. - W definicji funkcji dołącz argument o typie
fn.FabricVariablesClient. Ten klient udostępnia metody potrzebne do pracy z elementem biblioteki zmiennych. - Użyj
getVariables()metody , aby pobrać wszystkie zmienne z biblioteki zmiennych. - Aby odczytać wartości zmiennych, użyj polecenia
["variable-name"]lub.get("variable-name").
Przykład
W tym przykładzie symulujemy scenariusz konfiguracji dla środowiska produkcyjnego i programistycznego. Ta funkcja ustawia ścieżkę magazynowania w zależności od wybranego środowiska przy użyciu wartości pobranej z biblioteki zmiennych. Biblioteka zmiennych zawiera zmienną o nazwie ENV , w której użytkownicy mogą ustawić wartość dev lub 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"
Treści powiązane
- Dokumentacja interfejsu API referencyjnego
- Utwórz element funkcji danych użytkownika Fabric
- przykłady funkcji danych użytkownika