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

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

사용자 데이터 함수 SDK

사용자 데이터 함수 항목에는 제공된 REST 엔드포인트를 사용하여 Fabric 포털, 다른 Fabric 항목 또는 외부 애플리케이션에서 호출할 수 있는 하나 이상의 함수가 포함되어 있습니다. 각 함수는 매개 변수를 전달하고 호출자에게 출력을 반환할 수 있는 Python 스크립트의 메서드입니다. 사용자 데이터 함수 프로그래밍 모델에는 다음과 같은 구성 요소가 포함됩니다.

fabric.functions 라이브러리는 Python에서 사용자 데이터 함수를 만드는 데 필요한 코드를 제공합니다. 새로운 사용자 데이터 함수 항목을 만들면 첫 번째 함수 템플릿에 이 라이브러리가 가져오는 것을 볼 수 있습니다.
이 메서드 fn.UserDataFunctions() 는 모든 새 사용자 데이터 함수 항목에서 코드 파일의 시작 부분에 있는 실행 컨텍스트를 함수 정의 앞에 제공합니다.

예제:
```
import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()
```
모든 함수는 @udf.function() 데코레이터로 식별됩니다. 이 데코레이터는 포털 또는 외부 호출자에서 함수를 개별적으로 호출할 수 있는지 여부를 정의합니다. 또한 이 데코레이터를 사용하려면 함수에 반환 값이 있어야 합니다. 이 데코레이터를 사용하는 함수는 데코레이터가 나타내는 연결 개체에 @udf.connection 액세스할 수 있습니다.

호출 가능한 함수 예
```
# This is a hello fabric function sample that can be invoked from the Fabric portal, another Fabric item, or an external application.

@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 메서드는 직접 호출할 수 없습니다. 데코레이터를 포함하는 함수에서만 호출할 수 있으며, 도우미 함수로 사용할 수 있습니다.

도우미 함수 예
```
# This is a helper function that can be invoked from other functions, but can't be invoked or run directly because it doesn't have the @udf.function() decorator

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 형식을 사용하려면 사용자 데이터 함수에 대한 패브릭 포털에서 라이브러리 관리를 선택하고 버전을 1.0.0으로 업데이트 fabric-user-data-function 합니다.

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

{
  "name": "Alice",                          // String (str)
  "signup_date": "2025-07-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-07-01": {"product": "A", "units": 10},
    "2025-07-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

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

코드에서 함수 정의를 사용하여 비동기 데코레이터를 추가합니다. 함수를 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 데이터 원본에 대한 데이터 연결

이 모듈을 사용하면 코드에서 연결 문자열을 작성하지 않고도 데이터 연결을 참조할 수 있습니다. 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:
  # Replace with the query you want to run
  query = "SELECT * FROM (VALUES ('John Smith', 31), ('Kayla Jones', 33)) AS Employee(EmpName, DepID);"

  # [...] Here is where the rest of your SqlConnection code would be.

  return results

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

일반 연결을 사용하면 사용자 데이터 함수 항목 소유자 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 항목 에 연결할 수 있습니다.

패브릭 사용자 데이터 함수 항목에서 라이브러리 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를 호출하는 클라이언트 비밀을 검색할 수 있습니다.

패브릭 사용자 데이터 함수 항목에서 requests 사용하여 azure-keyvault-secrets 및 라이브러리를 설치합니다.
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}"

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

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

프로그래밍 모델에는 UserDataFunctionContext 개체도 포함됩니다. 이 개체는 함수 호출 메타데이터를 포함하고 있으며 특정 호출 메커니즘에 대한 특정 앱 논리를 만드는 데 사용할 수 있습니다.

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

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

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

속성 이름	데이터 형식	설명
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 = {context.executing_user['Oid']}, TenantId = {context.executing_user['TenantId']}, PreferredUsername = {context.executing_user['PreferredUsername']}, InvocationId = {context.invocation_id}"

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

함수를 개발할 때 Python 프로그래밍 모델에서 사용 가능한 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의 패브릭 변수 라이브러리 는 작업 영역 내의 여러 항목에서 사용할 수 있는 변수를 관리하기 위한 중앙 집중식 리포지토리입니다. 이를 통해 개발자는 항목 구성을 효율적으로 사용자 지정하고 공유할 수 있습니다.

연결 관리를 사용하여 변수 라이브러리에 연결을 추가하고 변수 라이브러리 항목의 별칭을 가져옵니다.
연결 데코레이터 @udf.connection(argName="varLib", alias="<My Variable Library Alias>") 를 추가하여 변수 라이브러리 항목의 별칭을 참조합니다.
함수 정의에 형식 fn.FabricVariablesClient이 있는 인수를 포함합니다. 이 클라이언트는 변수 라이브러리 항목을 사용하는 데 필요한 메서드를 제공합니다. 예를 들어 def standardize_date(rawDate: str, varLib: fn.FabricVariablesClient) -> str:
메서드를 사용하여 getVariables() 변수 라이브러리에서 모든 변수를 가져옵니다.

    # Get all variables from the variable library item
    variables = varLib.getVariables()

1 변수의 값을 읽으려면 사용 ["variable-name"] 하거나 .get("variable-name").

  # Get desired format from environment or use default
  date_format = variables["DATE_FORMAT"]
  # Another way to get the variable
  # date_format= variables.get("DATE_FORMAT")

피드백

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

Last updated on 2025-11-10