Azure OpenAI를 사용한 빅 데이터

Azure OpenAI 서비스를 사용하여 완료 API를 표시하여 많은 자연어 작업을 해결할 수 있습니다. 프롬프트 워크플로를 몇 가지 예제에서 예제의 큰 데이터 세트로 쉽게 확장할 수 있도록 Azure OpenAI 서비스는 분산 기계 학습 라이브러리 SynapseML 통합됩니다. 이 통합을 사용하면 Apache Spark 분산 컴퓨팅 프레임워크를 사용하여 OpenAI 서비스에서 수백만 개의 프롬프트를 처리할 수 있습니다. 이 자습서에서는 Azure OpenAI 및 Microsoft Fabric 사용하여 대규모 언어 모델을 분산된 규모로 적용하는 방법을 보여 줍니다.

필수 조건

이 빠른 시작의 주요 필수 구성 요소에는 작동 중인 Azure OpenAI 리소스 및 SynapseML이 설치된 Apache Spark 클러스터가 포함됩니다.

Microsoft Fabric 구독을 받으세요. 또는 무료 Microsoft Fabric 평가판 등록합니다.
Microsoft Fabric 로그인합니다.
홈 페이지의 왼쪽 아래에 있는 환경 전환기를 사용하여 패브릭으로 전환합니다.

Microsoft Fabric의 데이터 사이언스 업무 부대로 가세요.
새 Notebook을 만듭니다.
Azure OpenAI 리소스 - 리소스 만들기

이 가이드를 Notebook으로 가져오기

다음 단계에서는 이 코드를 Spark 클러스터에 추가합니다. Spark 플랫폼에서 Notebook을 만들고 코드를 이 Notebook에 복사하여 데모를 실행할 수 있습니다.

이 데모를 노트북으로 다운로드 하세요( Raw를 선택한 후 파일을 저장하세요).
Fabric 작업 공간으로 가져오세요.
설치 가이드를 사용 해 SynapseML을 클러스터에 설치하세요. 이 단계를 수행하려면 가져온 전자 필기장의 맨 위에 추가 셀을 붙여넣습니다.
Notebook을 클러스터에 연결하고 셀을 편집하고 실행하면서 따라해 봅니다.

서비스 정보 채우기

다음으로, 서비스를 가리키도록 Notebook의 셀을 편집합니다. service_nameOpenAI 서비스와 일치하도록 , deployment_name, location및 key 변수를 설정합니다.

import os
from pyspark.sql import SparkSession
from synapse.ml.core.platform import running_on_synapse, find_secret

# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()

if running_on_synapse():
    from notebookutils.visualization import display

# Fill in the following lines with your service information
# Learn more about selecting which embedding model to choose: https://openai.com/blog/new-and-improved-embedding-model
service_name = "synapseml-openai"
deployment_name = "gpt-4.1-mini"
deployment_name_embeddings = "text-embedding-3-small"

key = find_secret(
    "openai-api-key"
)  # please replace this line with your key as a string

assert key is not None and service_name is not None

프롬프트의 데이터 세트 만들기

다음으로, 각 행마다 하나의 프롬프트가 있는 일련의 행으로 구성된 데이터프레임을 만든다.

ADLS 또는 다른 데이터베이스에서 직접 데이터를 로드할 수도 있습니다. Spark DataFrame의 로드 및 준비에 대한 자세한 내용은 Apache Spark 데이터 로딩 가이드를 참조하세요.

df = spark.createDataFrame(
    [
        ("Hello my name is",),
        ("The best code is code that's",),
        ("SynapseML is ",),
    ]
).toDF("prompt")

OpenAIPrompt Apache Spark 클라이언트 만들기

Azure OpenAI 서비스를 DataFrame에 적용하려면, 분산 클라이언트 역할을 하는 객체를 OpenAIPrompt 생성하세요. 객체에 OpenAIPrompt 적절한 세터를 사용하여 단일 값 또는 DataFrame 컬럼으로 서비스 매개변수를 설정할 수 있습니다. 이 예제에서는 200으로 설정합니다 maxTokens . 토큰은 약 4자이며 이 제한은 프롬프트와 결과의 합계에 적용됩니다. DataFrame의 프롬프트 열명으로 매개변수를 promptCol 설정하세요.

from synapse.ml.services.openai import OpenAIPrompt

completion = (
    OpenAIPrompt()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name)
    .setCustomServiceName(service_name)
    .setMaxTokens(200)
    .setPromptCol("prompt")
    .setErrorCol("error")
    .setOutputCol("completions")
)

OpenAIPrompt 클라이언트를 사용하여 데이터프레임을 변환하세요

DataFrame과 프롬프트 클라이언트를 만든 후, 입력 데이터셋을 변환하고 서비스가 추가하는 모든 정보를 담은 열 completions 을 추가하세요. 간단히 하기 위해 텍스트만 선택합니다.

from pyspark.sql.functions import col

completed_df = completion.transform(df).cache()
display(
    completed_df.select(
        col("prompt"),
        col("error"),
        col("completions.choices.text").getItem(0).alias("text"),
    )
)

출력은 다음과 같이 표시됩니다. 완성 텍스트는 샘플과 다릅니다.

프롬프트	오류	text
안녕하세요, 제 이름은	null	Makaveli 나는 18 세이고 나는 로스 앤젤레스, CA에서 음악을 쓰고 만드는 것을 좋아 자랄 때 래퍼가되고 싶습니다.
가장 좋은 코드는 다음과 같은 코드입니다.	null	understandable 이것은 주관적인 진술이며 명확한 대답은 없습니다.
SynapseML은	null	이벤트의 향후 결과를 예측하는 방법을 배울 수 있는 기계 학습 알고리즘입니다.

추가 사용 예제

텍스트 임베딩 생성

텍스트를 완료하는 것 외에도 다운스트림 알고리즘 또는 벡터 검색 아키텍처에 사용할 텍스트를 포함할 수도 있습니다. 임베딩을 만들면 대규모 컬렉션에서 문서를 검색하고 검색할 수 있습니다. 프롬프트 엔지니어링이 작업에 충분하지 않은 경우 이 방법을 사용합니다. 사용에 OpenAIEmbedding대한 자세한 내용은 포함 가이드를 참조하세요.

from synapse.ml.services.openai import OpenAIEmbedding

embedding = (
    OpenAIEmbedding()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name_embeddings)
    .setCustomServiceName(service_name)
    .setTextCol("prompt")
    .setErrorCol("error")
    .setOutputCol("embeddings")
)

display(embedding.transform(df))

채팅 완료

GPT-4o 및 GPT-4.1과 같은 모델은 단일 프롬프트 대신 채팅을 이해합니다. OpenAIChatCompletion 변환기는 이 기능을 대규모로 공개합니다.

from synapse.ml.services.openai import OpenAIChatCompletion
from pyspark.sql import Row
from pyspark.sql.types import *


def make_message(role, content):
    return Row(role=role, content=content, name=role)


chat_df = spark.createDataFrame(
    [
        (
            [
                make_message(
                    "system", "You are an AI chatbot with red as your favorite color"
                ),
                make_message("user", "What's your favorite color"),
            ],
        ),
        (
            [
                make_message("system", "You are very excited"),
                make_message("user", "How are you today"),
            ],
        ),
    ]
).toDF("messages")


chat_completion = (
    OpenAIChatCompletion()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name)
    .setCustomServiceName(service_name)
    .setMessagesCol("messages")
    .setErrorCol("error")
    .setOutputCol("chat_completions")
)

display(
    chat_completion.transform(chat_df).select(
        "messages", "chat_completions.choices.message.content"
    )
)

요청 일괄 처리를 사용하여 처리량 향상

예제에서는 각 프롬프트에 대해 하나씩 서비스에 대한 여러 요청을 만듭니다. 단일 요청으로 여러 프롬프트를 완료하려면 일괄 처리 모드를 사용합니다. 먼저 개체에서 OpenAIPrompt 프롬프트 열을 "Prompt"로 설정하는 대신 BatchPrompt 열에 대해 "batchPrompt"를 지정합니다. 각 행에 프롬프트 목록을 포함하는 DataFrame을 만드세요.

batch_df = spark.createDataFrame(
    [
        (["The time has come", "Pleased to", "Today stocks", "Here's to"],),
        (["The only thing", "Ask not what", "Every litter", "I am"],),
    ]
).toDF("batchPrompt")

다음으로 개체를 만듭니다 OpenAIPrompt . 프롬프트 열을 설정하는 대신 열이 형식 Array[String]인 경우 batchPrompt 열을 설정합니다.

batch_completion = (
    OpenAIPrompt()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name)
    .setCustomServiceName(service_name)
    .setMaxTokens(200)
    .setBatchPromptCol("batchPrompt")
    .setErrorCol("error")
    .setOutputCol("completions")
)

변환 호출에서 행별로 요청이 이루어집니다. 각 행에는 여러 프롬프트가 포함되어 있으므로 각 요청은 해당 행의 모든 프롬프트를 보냅니다. 결과에는 요청의 각 행에 대한 행이 포함됩니다.

completed_batch_df = batch_completion.transform(batch_df).cache()
display(completed_batch_df)

자동 미니 배처 사용

데이터가 열 형식인 경우 SynapseML FixedMiniBatcherTransformer을 사용하여 데이터를 행 형식으로 변환할 수 있습니다.

from pyspark.sql.types import StringType
from synapse.ml.stages import FixedMiniBatchTransformer
from synapse.ml.core.spark import FluentAPI

completed_autobatch_df = (
    df.coalesce(
        1
    )  # Force a single partition so that our little 4-row DataFrame makes a batch of size 4, you can remove this step for large datasets
    .mlTransform(FixedMiniBatchTransformer(batchSize=4))
    .withColumnRenamed("prompt", "batchPrompt")
    .mlTransform(batch_completion)
)

display(completed_autobatch_df)

번역을 위한 프롬프트 엔지니어링 기술

Azure OpenAI 서비스는 프롬프트 엔지니어링 통해 다양한 자연어 작업을 해결할 수 있습니다. 이 예제에서는 언어 번역을 묻는 메시지를 표시합니다.

translate_df = spark.createDataFrame(
    [
        ("Japanese: Ookina hako \nEnglish: Big box \nJapanese: Midori tako\nEnglish:",),
        (
            "French: Quel heure et il au Montreal? \nEnglish: What time is it in Montreal? \nFrench: Ou est le poulet? \nEnglish:",
        ),
    ]
).toDF("prompt")

display(completion.transform(translate_df))

질문 답변 프롬프트

이 예제에서는 모델에게 일반 지식 질문에 답변하도록 메시지를 보여줍니다.

qa_df = spark.createDataFrame(
    [
        (
            "Q: Where is the Grand Canyon?\nA: The Grand Canyon is in Arizona.\n\nQ: What is the weight of the Burj Khalifa in kilograms?\nA:",
        )
    ]
).toDF("prompt")

display(completion.transform(qa_df))

피드백

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

Last updated on 2026-08-21