패브릭 사용자 데이터 함수 프로그래밍 모델 개요

Fabric 사용자 데이터 함수 프로그래밍 모델은 패브릭에서 함수를 작성하기 위한 패턴과 개념을 정의합니다.

SDK는 fabric-user-data-functions 이 프로그래밍 모델을 구현하여 실행 가능한 함수를 작성하고 게시하는 데 필요한 기능을 제공합니다. SDK를 사용하면 Fabric 데이터 원본 등 Fabric 에코시스템의 다른 항목과 원활하게 통합할 수도 있습니다. 이 라이브러리는 PyPI에서 공개적으로 사용 가능하며 사용자 데이터 함수 항목에 미리 설치되어 있습니다.

이 문서에서는 SDK를 사용하여 REST API를 사용하여 패브릭 포털, 다른 패브릭 항목 또는 외부 애플리케이션에서 호출할 수 있는 함수를 빌드하는 방법을 설명합니다. 실제 예제를 사용하여 프로그래밍 모델 및 주요 개념을 알아봅니다.

팁 (조언)

모든 클래스, 메서드 및 매개 변수에 대한 자세한 내용은 SDK 참조 설명서를 참조하세요.

SDK 시작

이 섹션에서는 사용자 데이터 함수 SDK의 핵심 구성 요소를 소개하고 함수를 구성하는 방법을 설명합니다. 필요한 가져오기, 데코레이터 및 함수에서 처리할 수 있는 입력 및 출력 데이터의 형식에 대해 알아봅니다.

사용자 데이터 함수 SDK

SDK는 fabric-user-data-functions Python에서 사용자 데이터 함수를 만드는 데 필요한 핵심 구성 요소를 제공합니다.

필수 가져오기 및 초기화

모든 사용자 데이터 함수 파일은 모듈을 fabric.functions 가져오고 실행 컨텍스트를 초기화해야 합니다.

import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()

@udf.function() 데코레이터

데코레이터로 표시된 함수는 @udf.function() 패브릭 포털, 다른 패브릭 항목 또는 외부 애플리케이션에서 호출할 수 있습니다. 이 데코레이터를 사용하는 함수는 반환 형식을 지정해야 합니다.

예제:

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

도우미 함수

데코레이터가 @udf.function() 없는 Python 메서드는 직접 호출할 수 없습니다. 데코레이팅된 함수에서만 호출할 수 있으며 도우미 함수로 사용할 수 있습니다.

예제:

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

지원되는 입력 유형

str, int, float 등과 같은 기본 데이터 형식과 같은 함수의 입력 매개 변수를 정의할 수 있습니다. 지원되는 입력 데이터 형식은 다음과 같습니다.

JSON 형식	Python 데이터 형식
문자열	스트링
DateTime 문자열	날짜 및 시간
Boolean	부울
숫자	int(정수형), float(부동소수점형)
배열	리스트[], 예시 리스트[int]
객체	사전
객체	pandas DataFrame
개체 또는 개체 배열	pandas 시리즈

비고

pandas DataFrame 및 Series 형식을 사용하려면 Fabric 포털로 이동하여 작업 영역을 찾고 사용자 데이터 함수 항목을 엽니다. 라이브러리 관리를 선택하고, 패키지를 검색 fabric-user-data-functions 하고, 버전 1.0.0 이상으로 업데이트합니다.

지원되는 입력 형식에 대한 요청 본문의 예:

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

지원되는 출력 형식

지원되는 출력 데이터 형식은 다음과 같습니다.

Python 데이터 형식
스트링
날짜 및 시간
부울
int(정수형), float(부동소수점형)
list[데이터 형식], 예 list[int]
사전
없음
pandas 시리즈
pandas DataFrame

함수 작성

구문 요구 사항 및 제한 사항

사용자 데이터 함수를 작성할 때 함수가 제대로 작동하도록 특정 구문 규칙을 따라야 합니다.

매개 변수 이름 지정

camelCase 사용: 매개 변수 이름은 camelCase 명명 규칙을 사용해야 하며 밑줄을 포함할 수 없습니다. 예를 들어 productName 대신 product_name을 사용합니다.
예약된 키워드: 예약된 Python 키워드 또는 다음 Fabric 관련 키워드를 매개 변수 이름 또는 함수 이름으로 reqcontextreqInvocationId사용할 수 없습니다.

매개 변수 요구 사항

기본값 없음: 기본 매개 변수 값은 지원되지 않습니다. 함수를 호출할 때는 모든 매개 변수가 필요합니다. 예를 들어, 다음 함수는 구문 오류를 발생시킵니다.
```
# 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}."
```
형식 주석 필요: 모든 매개 변수에는 형식 주석(예: name: str)이 포함되어야 합니다.

함수 요구 사항

반환 형식 필요: 데코레이터가 @udf.function() 있는 함수는 반환 형식 주석(예: -> str)을 지정해야 합니다.
필수 임포트: 함수가 작동하려면 import fabric.functions as fn 문장 및 udf = fn.UserDataFunctions() 초기화가 필요합니다.

올바른 구문의 예

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

비동기 함수를 작성하는 방법

코드에서 함수 정의를 사용하여 비동기 데코레이터를 추가합니다. 함수를 async 사용하면 여러 작업을 한 번에 처리하여 애플리케이션의 응답성과 효율성을 향상시킬 수 있습니다. 대용량 I/O 바인딩된 작업을 관리하는 데 적합합니다. 이 예제 함수는 pandas를 사용하여 레이크하우스에서 CSV 파일을 읽습니다. 함수는 파일 이름을 입력 매개 변수로 사용합니다.

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

데이터 작업

Fabric 데이터 원본에 대한 데이터 연결

SDK를 사용하면 코드에 연결 문자열 을 작성할 필요 없이 데이터 연결을 참조할 수 있습니다. fabric.functions 라이브러리는 데이터 연결을 처리하는 두 가지 방법을 제공합니다.

fabric.functions.FabricSqlConnection: SQL Analytics 엔드포인트 및 Fabric 웨어하우스를 포함하여 Fabric에서 SQL 데이터베이스로 작업할 수 있습니다.
fabric.functions.FabricLakehouseClient: 레이크하우스 테이블과 레이크하우스 파일에 모두 연결할 수 있는 방법을 통해 레이크하우스와 함께 작업할 수 있습니다.

데이터 원본에 대한 연결을 참조하려면 @udf.connection 데코레이터를 사용해야 합니다. 다음 형식 중 하나로 적용할 수 있습니다.

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

@udf.connection에 대한 인수는 다음과 같습니다.

argName- 연결이 함수에 사용하는 변수의 이름입니다.
alias는 연결 관리 메뉴로 추가한 연결의 별칭입니다.
argName과 alias의 값이 같으면 @udf.connection("<alias and argName for the data connection>")을 사용할 수 있습니다.

예시

# 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

패브릭 항목 또는 Azure 리소스에 대한 일반 연결

SDK는 사용자 데이터 함수 항목 소유자 ID를 사용하여 패브릭 항목 또는 Azure 리소스에 대한 연결을 만들 수 있는 일반 연결을 지원합니다. 이 기능은 항목 소유자의 ID 및 제공된 대상 그룹 유형을 사용하여 Microsoft Entra ID 토큰을 생성합니다. 이 토큰은 해당 대상 그룹 유형을 지원하는 패브릭 항목 또는 Azure 리소스로 인증하는 데 사용됩니다. 이 방법은 연결 관리 기능 에서 관리되는 연결 개체를 사용하는 것과 비슷한 프로그래밍 환경을 제공하지만 연결에서 제공된 대상 그룹 유형에 대해서만 사용할 수 있습니다.

이 기능은 다음 매개 변수와 함께 데코레이터를 사용합니다 @udf.generic_connection() .

매개 변수	설명	가치
`argName`	함수에 전달되는 변수의 이름입니다. 사용자는 함수의 인수에서 이 변수를 지정하고 해당 변수의 `fn.FabricItem` 형식을 사용해야 합니다.	만약 `argName=CosmosDb` 경우에는, 함수에 이 인수 `cosmosDb: fn.FabricItem`가 포함되어야 합니다.
`audienceType`	연결이 만들어지는 대상 그룹의 유형입니다. 이 매개 변수는 패브릭 항목 또는 Azure 서비스의 형식과 연결되며 연결에 사용되는 클라이언트를 결정합니다.	이 매개 변수에 허용되는 값은 다음과 `CosmosDb`같습니다`KeyVault`.

일반 연결을 사용하여 Fabric Cosmos DB 컨테이너에 연결

일반 연결은 CosmosDB 오디언스 유형을 사용하여 네이티브 Fabric Cosmos DB 항목을 지원합니다. 포함된 사용자 데이터 함수 SDK는 모든 호출에 대해 단일 Cosmos DB 클라이언트를 가져오는 도우미 get_cosmos_client 메서드를 제공합니다.

다음 단계에 따라 제네릭 연결을 사용하여 Fabric Cosmos DB 항목 에 연결할 수 있습니다.

Fabric 포털로 이동하여 작업 영역을 찾고 사용자 데이터 함수 항목을 엽니다. 라이브러리 관리를 선택하고, 라이브러리를 azure-cosmos 검색하고, 설치합니다. 자세한 내용은 라이브러리 관리를 참조하세요.
Fabric Cosmos DB 항목 설정으로 이동합니다.
Fabric Cosmos DB 엔드포인트 URL을 검색합니다.

사용자 데이터 함수 항목으로 이동합니다. 다음 샘플 코드를 사용하여 Fabric Cosmos DB 컨테이너에 연결하고 Cosmos DB 샘플 데이터 세트를 사용하여 읽기 쿼리를 실행합니다. 다음 변수의 값을 바꿉니다.

COSMOS_DB_URI Fabric Cosmos DB 엔드포인트와 함께 사용하십시오.
DB_NAME 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)

이 함수를 테스트하거나 실행하려면 호출 매개 변수로 Accessory을 제공하세요.

비고

이러한 단계를 사용하여 계정 URL 및 데이터베이스 이름을 사용하여 Azure Cosmos DB 데이터베이스에 연결할 수도 있습니다. 사용자 데이터 함수 소유자 계정에는 해당 Azure Cosmos DB 계정에 대한 액세스 권한이 필요합니다.

일반 연결을 사용하여 Azure Key Vault에 연결

일반 연결은 대상 그룹 유형을 사용하여 Azure Key Vault에 연결할 수 KeyVault 있도록 지원합니다. 이러한 유형의 연결을 사용하려면 Fabric 사용자 데이터 함수 소유자가 Azure Key Vault에 연결할 수 있는 권한이 있어야 합니다. 이 연결을 사용하여 이름으로 키, 비밀 또는 인증서를 검색할 수 있습니다.

다음 단계를 수행하여 Azure Key Vault 에 연결하여 일반 연결을 사용하여 API를 호출하는 클라이언트 비밀을 검색할 수 있습니다.

Fabric 포털로 이동하여 작업 영역을 찾고 사용자 데이터 함수 항목을 엽니다. 라이브러리 관리를 선택한 다음, 라이브러리를 검색하여 requests 설치 azure-keyvault-secrets 합니다. 자세한 내용은 라이브러리 관리를 참조하세요.
Azure 포털에서 Azure Key Vault 리소스로 이동하여 키, 비밀 또는 인증서의 Vault URI와 이름을 검색합니다.

패브릭 사용자 데이터 함수 항목으로 돌아가서 이 샘플을 사용합니다. 이 샘플에서는 Azure Key Vault에서 비밀을 검색하여 공용 API에 연결합니다. 다음 변수의 값을 바꿉니다.

KEY_VAULT_URL 이전 단계에서 검색한 Vault URI와 함께
KEY_VAULT_SECRET_NAME 비밀의 이름으로 지정합니다.
API_URL 연결하려는 API의 URL이 있는 변수입니다. 이 샘플에서는 GET 요청을 수락하고 다음 매개 변수 api-key 를 사용하는 공용 API에 연결한다고 가정합니다 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}"

코드에 요청 본문을 제공하여 이 함수를 테스트하거나 실행해 보세요.

고급 기능

프로그래밍 모델은 함수를 보다 세세하게 제어할 수 있는 고급 패턴을 정의합니다. SDK는 다음을 수행할 수 있는 클래스 및 메서드를 통해 이러한 패턴을 구현합니다.

함수를 호출한 사용자 및 방법에 대한 호출 메타데이터에 액세스
구조적 오류 응답을 사용하여 사용자 지정 오류 시나리오 처리
중앙 집중식 구성 관리를 위해 패브릭 변수 라이브러리와 통합

비고

사용자 데이터 함수에는 요청 크기, 실행 시간 제한 및 응답 크기에 대한 서비스 제한이 있습니다. 이러한 제한 및 적용 방법에 대한 자세한 내용은 서비스 세부 정보 및 제한 사항을 참조하세요.

`UserDataFunctionContext`를 사용하여 호출 속성 가져오기

SDK에는 개체가 UserDataFunctionContext 포함됩니다. 이 개체는 함수 호출 메타데이터를 포함하며 다양한 호출 메커니즘(예: 포털 호출 및 REST API 호출)에 대한 특정 앱 논리를 만드는 데 사용할 수 있습니다.

다음 표는 UserDataFunctionContext 개체의 속성을 보여 줍니다.

속성 이름	데이터 형식	설명
invocation_id	문자열	사용자 데이터 함수 항목 호출과 연결된 고유한 GUID입니다.
executing_user	객체	호출을 권한 부여하는 데 사용되는 사용자 정보의 메타데이터입니다.

executing_user 개체에는 다음 정보가 포함되어 있습니다.

속성 이름	데이터 형식	설명
OID	문자열(GUID)	요청자에 대한 변경 불가능한 식별자인 사용자의 개체 ID입니다. 이는 여러 애플리케이션에서 이 함수를 호출하는 데 사용되는 사용자 또는 서비스 주체의 검증된 ID입니다.
테넌트 ID	문자열(GUID)	사용자가 로그인한 테넌트의 ID입니다.
기본 사용자 이름	문자열	사용자가 설정한 호출 사용자의 기본 사용자 이름입니다. 이 값은 변경 가능합니다.

UserDataFunctionContext 매개 변수에 액세스하려면 함수 정의 맨 위에 다음 데코레이터를 사용해야 합니다. @udf.context(argName="<parameter name>")

예시

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

`UserThrownError`로 처리된 오류를 throw합니다.

함수를 개발할 때 SDK에서 사용할 수 있는 클래스를 UserThrownError 사용하여 예상되는 오류 응답을 throw할 수 있습니다. 이 클래스의 한 가지 용도는 사용자가 제공한 입력이 비즈니스 유효성 검사 규칙을 통과하지 못하는 경우를 관리하는 것입니다.

예시

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

UserThrownError 클래스 생성자는 두 개의 매개 변수를 사용합니다.

Message: 이 문자열은 이 함수를 호출하는 애플리케이션에 대한 오류 메시지로 반환됩니다.
속성 사전이 이 함수를 호출하는 애플리케이션에 반환됩니다.

패브릭 변수 라이브러리에서 변수 가져오기

Microsoft Fabric의 패브릭 변수 라이브러리 는 작업 영역 내의 여러 항목에서 사용할 수 있는 변수를 관리하기 위한 중앙 집중식 리포지토리입니다. 이를 통해 개발자는 항목 구성을 효율적으로 사용자 지정하고 공유할 수 있습니다. 아직 변수 라이브러리가 없는 경우 변수 라이브러리 만들기 및 관리를 참조하세요.

함수에서 변수 라이브러리를 사용하려면 사용자 데이터 함수 항목에서 해당 라이브러리에 대한 연결을 추가합니다. 변수 라이브러리는 SQL 데이터베이스 및 레이크하우스와 같은 데이터 원본과 함께 OneLake 카탈로그에 표시됩니다.

함수에서 변수 라이브러리를 사용하려면 다음 단계를 수행합니다.

사용자 데이터 함수 항목에서 변수 라이브러리 에 대한 연결을 추가 합니다. OneLake 카탈로그에서 변수 라이브러리를 찾아 선택한 다음, 연결을 선택합니다. 패브릭이 연결에 대해 생성하는 별칭 을 확인합니다.
변수 라이브러리 항목에 대한 연결 데코레이터를 추가합니다. 예를 들어 @udf.connection(argName="varLib", alias="<My Variable Library Alias>") 별칭을 변수 라이브러리 항목에 대해 새로 추가된 연결로 바꿉니다.
함수 정의에 형식 fn.FabricVariablesClient이 있는 인수를 포함합니다. 이 클라이언트는 변수 라이브러리 항목을 사용하는 데 필요한 메서드를 제공합니다.
메서드를 사용하여 getVariables() 변수 라이브러리에서 모든 변수를 가져옵니다.
변수의 값을 읽으려면 다음 중 하나 ["variable-name"] 또는 .get("variable-name")를 사용합니다.

예시

이 예제에서는 프로덕션 및 개발 환경에 대한 구성 시나리오를 시뮬레이션합니다. 이 함수는 변수 라이브러리에서 검색된 값을 사용하여 선택한 환경에 따라 스토리지 경로를 설정합니다. 변수 라이브러리에는 사용자가 값을 ENV으로 또는 dev로 설정할 수 있는 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"

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-01-22