如何在Azure Data Factory中调用Python REST客户端并将结果存入Azure SQL仓库？

Got it, let's break down exactly how to make this work for you. Since you already have your Python REST client sorted with OAuth and token handling, we just need to wire it into Azure Data Factory (ADF) and get those API results into your Azure SQL Warehouse. Here's a step-by-step guide:

First, wrap your existing REST logic into a script that either outputs data to a storage location ADF can access, or directly writes to your SQL Warehouse. Here's a sample with both options:

import requests
import pandas as pd
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential

# Reuse your existing OAuth token logic
def fetch_oauth_token():
    # Your pre-configured token retrieval code here
    # Pro tip: Store client IDs/secrets in Azure Key Vault instead of hardcoding!
    vault_client = SecretClient(
        vault_url="https://your-keyvault.vault.azure.net/",
        credential=DefaultAzureCredential()
    )
    client_secret = vault_client.get_secret("api-oauth-secret").value
    # ... rest of your token fetch logic ...
    return "valid_access_token"

def call_target_api():
    token = fetch_oauth_token()
    headers = {"Authorization": f"Bearer {token}"}
    api_response = requests.get("https://your-api-endpoint.com/data", headers=headers)
    api_response.raise_for_status() # Fail fast on API errors
    return api_response.json()

def main():
    api_data = call_target_api()
    df = pd.DataFrame(api_data)

    # Option 1: Output to Azure Blob Storage (for ADF Copy Activity to pick up)
    df.to_json("/mnt/adf-blob/output/api_results.json", orient="records")

    # Option 2: Directly write to Azure SQL Warehouse (skip ADF Copy Activity)
    # conn_str = SecretClient(...).get_secret("sql-connection-string").value
    # df.to_sql(
    #     name="target_table",
    #     con=conn_str,
    #     if_exists="append",
    #     index=False
    # )

if __name__ == "__main__":
    main()

• Use Option 1 if you want ADF to handle data transformation/validation before writing to SQL.
• Use Option 2 for a more direct flow, ideal if you don't need extra processing steps.

You have two reliable options here, depending on your needs:

Option A: Self-Hosted Integration Runtime (Self-Hosted IR) + Custom Activity #

Great if your script has niche dependencies or needs access to on-prem resources:

• Deploy a Self-Hosted IR to a VM (Azure or on-prem) that can reach your API and SQL Warehouse.
• Install Python, requests, pandas, and any other required libraries on the IR machine.
• In ADF, create a Custom Activity:
  • Link it to your Self-Hosted IR.
  • Point the script path to either the IR's local filesystem or a mounted Azure Blob container.
• If you used Option 1 in your script, add a Copy Activity after the Custom Activity to pull the JSON/CSV from Blob Storage into your SQL Warehouse.

Option B: Azure Function + ADF Web Activity #

Perfect for serverless, low-maintenance execution:

• Create an Azure Function (HTTP trigger recommended) and paste your Python script logic into it.
• List all dependencies in a requirements.txt file (e.g., requests, pandas, azure-keyvault-secrets).
• Secure the Function with ADF's managed identity: Add the ADF managed identity to the Function's Access Control (IAM) with "Function App Contributor" permissions.
• In ADF, add a Web Activity that calls the Function's HTTP endpoint. Use the managed identity for authentication.
• Again, use a Copy Activity if you're outputting to Blob Storage, or skip it if the Function writes directly to SQL.

Never hardcode secrets. Use Azure Key Vault for all sensitive data:

• Store OAuth client secrets, SQL connection strings, and API keys as secrets in Key Vault.
• Grant your Self-Hosted IR machine or Azure Function's managed identity access to the Key Vault (via IAM, with "Key Vault Secrets User" permissions).
• Use the DefaultAzureCredential class in Python to fetch secrets without hardcoding credentials.

• Test each activity individually first: Verify the script runs successfully, then check if data lands in Blob Storage/SQL.
• Use ADF's built-in monitoring dashboard to track pipeline runs and debug errors (e.g., API timeouts, SQL permission issues).
• For Self-Hosted IR, ensure the machine stays online or configure auto-start for reliability.

内容的提问来源于stack exchange，提问作者Gagan