Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
Data Factory in Microsoft Fabric bietet Cloud-Skalierungsdienste für Datenverschiebungen und Datentransformation, mit denen Sie die komplexesten Data Factory- und ETL-Szenarien lösen können, und die Ihnen eine moderne Datenintegrationserfahrung bieten, um Daten aus einer Vielzahl von Datenquellen aufzunehmen, vorzubereiten und zu transformieren. Innerhalb von Data Factory können Sie Datenpipelines erstellen, um sofort einsatzbereite umfangreiche Daten-Orchestrierungsfunktionen zu verwenden, um flexible Datenworkflows zu erstellen, die Ihren Unternehmensanforderungen entsprechen.
Mit dem KI-Modell gpt-4o in Azure erweitern wir nun die Möglichkeiten von dem, was Sie mit Data Factory tun können, und ermöglichen es Ihnen, Datenlösungen aus einem Bild zu erstellen.
Was Sie für die ersten Schritte benötigen Nur ein Microsoft Fabric-Konto und eine Idee. Hier zeigen wir Ihnen, wie Sie eine Whiteboardidee mithilfe eines Bilds und gpt-4o in eine Fabric Data Factory-Datenpipeline transformieren.
Voraussetzungen
Bevor Sie die Lösung erstellen, stellen Sie sicher, dass die folgenden Voraussetzungen in Azure und Fabric eingerichtet sind:
- Microsoft Fabric-fähiger Arbeitsbereich.
- Azure OpenAI-Konto mit einem API-Schlüssel und einem bereitgestellten
gpt-4o-Modell. - Ein Bild davon, wie Ihre Pipeline aussehen soll.
Warnung
API-Schlüssel sind vertrauliche Informationen und Produktionsschlüssel sollten immer nur sicher in Azure Key Vault oder anderen sicheren Speichern gespeichert werden. In diesem Beispiel wird für Demonstrationszwecke nur ein OpenAI-Schlüssel verwendet. Bei Produktionscode sollten Sie für eine sicherere Umgebung Microsoft Entra ID anstelle von Schlüsselauthentifizierung verwenden, weil sie nicht auf die Schlüsselfreigabe angewiesen ist und keine eine Sicherheitsverletzung riskiert, wenn ein Schlüssel kompromittiert wird.
Schritt 1: Hochladen Ihres Bilds in ein Lakehouse
Bevor Sie das Bild analysieren können, müssen Sie es in Ihr Lakehouse hochladen. Melden Sie sich bei Ihrem Microsoft Fabric-Konto an und navigieren Sie zu Ihrem Arbeitsbereich. Wählen Sie + Neues Element aus, und erstellen Sie ein neues Lakehouse.
Nachdem Ihr Lakehouse eingerichtet ist, erstellen Sie einen neuen Ordner unter Dateien, die als Bilder bezeichnet werden, und laden Sie das Bild dort hoch.
Schritt 2: Erstellen des Notebooks in Ihrem Arbeitsbereich
Jetzt müssen wir nur ein Notebook erstellen, um Python-Code auszuführen, der die Pipeline im Arbeitsbereich zusammenfasst und erstellt.
Erstellen Sie ein neues Notebook in Ihrem Arbeitsbereich:
Geben Sie im Codebereich den folgenden Code ein, der die erforderlichen Bibliotheken und Konfigurationen einrichtet und das Bild codiert:
# Configuration
AZURE_OPENAI_KEY = "<Your Azure OpenAI key>"
AZURE_OPENAI_GPT4O_ENDPOINT = "<Your Azure OpenAI gpt-4o deployment endpoint>"
IMAGE_PATH = "<Path to your uploaded image file>" # For example, "/lakehouse/default/files/images/pipeline.png"
# Install the OpenAI library
!pip install semantic-link --q
!pip uninstall --yes openai
!pip install openai
%pip install openai --upgrade
# Imports
import os
import requests
import base64
import json
import time
import pprint
import openai
import sempy.fabric as fabric
import pandas as pd
# Load the image
image_bytes = open(IMAGE_PATH, 'rb').read()
encoded_image = base64.b64encode(image_bytes).decode('ascii')
## Request headers
headers = {
"Content-Type": "application/json",
"api-key": AZURE_OPENAI_KEY,
}
Führen Sie diesen Codeblock aus, um die Umgebung zu konfigurieren.
Schritt 3: Verwenden von gpt-4o zum Beschreiben der Pipeline (optional)
Dieser Schritt ist optional, zeigt Ihnen aber, wie einfach es ist, Details aus dem Bild zu extrahieren, die für Ihre Zwecke relevant sein könnten. Wenn Sie diesen Schritt nicht ausführen, können Sie den JSON-Code der Pipeline im nächsten Schritt trotzdem generieren.
Wählen Sie zuerst Bearbeiten im Hauptmenü für das Notebook aus und wählen Sie dann die Schaltfläche + Codezelle unterhalb auf der Symbolleiste aus, um einen neuen Codeblock nach dem vorherigen hinzuzufügen.
Fügen Sie dann den folgenden Code zum neuen Abschnitt hinzu. Dieser Code zeigt, wie gpt-4o das Bild interpretieren und zusammenfassen kann, um dessen Inhalt zu verstehen.
# Summarize the image
## Request payload
payload = {
"messages": [
{
"role": "system",
"content": [
{
"type": "text",
"text": "You are an AI assistant that helps an Azure engineer understand an image that likely shows a Data Factory in Microsoft Fabric data pipeline. Show list of pipeline activities and how they are connected."
}
]
},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{encoded_image}"
}
}
]
}
],
"temperature": 0.7,
"top_p": 0.95,
"max_tokens": 800
}
## Send request
try:
response = requests.post(AZURE_OPENAI_GPT4O_ENDPOINT, headers=headers, json=payload)
response.raise_for_status() # Will raise an HTTPError if the HTTP request returned an unsuccessful status code
except requests.RequestException as e:
raise SystemExit(f"Failed to make the request. Error: {e}")
response_json = response.json()
## Show AI response
print(response_json["choices"][0]['message']['content'])
Führen Sie diesen Codeblock aus, um die KI-Zusammenfassung des Bilds und der zugehörigen Komponenten anzuzeigen.
Schritt 4: Generieren der Pipeline-JSON
Fügen Sie dem Notebook einen weiteren Codeblock hinzu und fügen Sie den folgenden Code hinzu. Dieser Code analysiert das Bild und generiert die Pipeline-JSON.
# Analyze the image and generate the pipeline JSON
## Setup new payload
payload = {
"messages": [
{
"role": "system",
"content": [
{
"type": "text",
"text": "You are an AI assistant that helps an Azure engineer understand an image that likely shows a Data Factory in Microsoft Fabric data pipeline. Succeeded is denoted by a green line, and Fail is denoted by a red line. Generate an ADF v2 pipeline JSON with what you see. Return ONLY the JSON text required, without any leading or trailing markdown denoting a code block."
}
]
},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{encoded_image}"
}
}
]
}
],
"temperature": 0.7,
"top_p": 0.95,
"max_tokens": 800
}
## Send request
try:
response = requests.post(AZURE_OPENAI_GPT4O_ENDPOINT, headers=headers, json=payload)
response.raise_for_status() # Will raise an HTTPError if the HTTP request returned an unsuccessful status code
except requests.RequestException as e:
raise SystemExit(f"Failed to make the request. Error: {e}")
## Get JSON from request and show
response_json = response.json()
pipeline_json = response_json["choices"][0]['message']['content']
print(pipeline_json)
Führen Sie diesen Codeblock aus, um die Pipeline-JSON aus dem Image zu generieren.
Schritt 4: Erstellen der Pipeline mit Fabric-REST-APIs
Nachdem Sie die Pipeline-JSON abgerufen haben, können Sie sie direkt mithilfe der Fabric-REST-APIs erstellen. Fügen Sie dem Notebook einen weiteren Codeblock hinzu und fügen Sie den folgenden Code hinzu. Dieser Code erstellt die Pipeline in Ihrem Arbeitsbereich.
# Convert pipeline JSON to Fabric REST API request
json_data = json.loads(pipeline_json)
# Extract the activities from the JSON
activities = json_data["properties"]["activities"]
# Prepare the data pipeline JSON definition
data = {}
activities_list = []
idx = 0
# Name mapping used to track activity name found in image to dynamically generated name
name_mapping = {}
for activity in activities:
idx = idx + 1
activity_name = activity["type"].replace("Activity","")
objName = f"{activity_name}{idx}"
# store the name mapping so we can deal with dependency
name_mapping[activity["name"]] = objName
if 'dependsOn' in activity:
activity_dependent_list = activity["dependsOn"]
dependent_activity = ""
if ( len(activity_dependent_list) > 0 ):
dependent_activity = activity_dependent_list[0]["activity"]
match activity_name:
case "Copy":
activities_list.append({'name': objName, 'type': "Copy", 'dependsOn': [],
'typeProperties': { "source": { "datasetSettings": {} },
"sink": { "datasetSettings": {} } }})
case "Web":
activities_list.append({'name': objName, 'type': "Office365Outlook",
"dependsOn": [
{
"activity": name_mapping[dependent_activity] ,
"dependencyConditions": [
"Succeeded"
]
}
]
}
)
case "ExecutePipeline":
activities_list.append({'name': "execute pipeline 1", 'type': "ExecutePipeline",
"dependsOn": [
{
"activity": name_mapping[dependent_activity] ,
"dependencyConditions": [
"Succeeded"
]
}
]
}
)
case _:
continue
else:
# simple activities with no dependencies
match activity_name:
case "Copy":
activities_list.append({'name': objName, 'type': "Copy", 'dependsOn': [],
'typeProperties': { "source": { "datasetSettings": {} } , "sink": { "datasetSettings": {} } }})
case "SendEmail":
activities_list.append({'name': "Send mail on success", 'type': "Office365Outlook"})
case "Web":
activities_list.append({'name': "Send mail on success", 'type': "Office365Outlook"})
case "ExecutePipeline":
activities_list.append({'name': "execute pipeline 1", 'type': "ExecutePipeline"})
case _:
print("NoOp")
# Now that the activities_list is created, assign it to the activities tag in properties
data['properties'] = { "activities": activities_list}
# Convert data from dict to string, then Byte Literal, before doing a Base-64 encoding
data_str = str(data).replace("'",'"')
createPipeline_json = data_str.encode(encoding="utf-8")
createPipeline_Json64 = base64.b64encode(createPipeline_json)
# Create a new data pipeline in Fabric
timestr = time.strftime("%Y%m%d-%H%M%S")
pipelineName = f"Pipeline from image with AI-{timestr}"
payload = {
"displayName": pipelineName,
"type": "DataPipeline",
"definition": {
"parts": [
{
"path": "pipeline-content.json",
"payload": createPipeline_Json64,
"payloadType": "InlineBase64"
}
]
}
}
print(f"Creating pipeline: {pipelineName}")
# Call the Fabric REST API to generate the pipeline
client = fabric.FabricRestClient()
workspaceId = fabric.get_workspace_id()
try:
response = client.post(f"/v1/workspaces/{workspaceId}/items",json=payload)
if response.status_code != 201:
raise FabricHTTPException(response)
except WorkspaceNotFoundException as e:
print("Workspace is not available or cannot be found.")
except FabricHTTPException as e:
print(e)
print("Fabric HTTP Exception. Check that you have the correct Fabrric API endpoints.")
response = client.get(f"/v1/workspaces/{workspaceId}/Datapipelines")
df_items = pd.json_normalize(response.json()['value'])
print("List of pipelines in the workspace:")
df_items
Die Ausgabe bestätigt den Namen der erstellten Pipeline und zeigt eine Liste der Pipelines Ihres Arbeitsbereichs an, sodass Sie überprüfen können, ob sie vorhanden ist.
Schritt 6: Verwenden der Pipeline
Nachdem Ihre Pipeline erstellt wurde, können Sie sie in Ihrem Fabric-Arbeitsbereich bearbeiten, um ihr Bild als Pipeline zu sehen. Sie können jede Aktivität auswählen, um sie wie gewünscht zu konfigurieren, und sie dann nach Bedarf ausführen und überwachen.
