在Azure Function中传递Azure Event Hubs分区键至Azure SQL存储过程

你当前配置报错的核心原因是：EventHub触发器设置为cardinality: "many"（批量接收事件）时，SQL输入绑定无法直接通过{PartitionKey}获取分区键。因为输入绑定在函数执行前就完成解析，无法动态关联批量中每个事件的元数据。

以下是两种可行的解决方案：

方案一：切换为单事件处理模式 #

将EventHub触发器改为单事件接收（cardinality: "one"），此时每个函数调用仅处理一个事件，SQL输入绑定可以直接引用该事件的分区键。

修改后的function.json #

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "event",
      "direction": "in",
      "eventHubName": "playerinput",
      "connection": "AzureEventHubConnectionString",
      "cardinality": "one",
      "consumerGroup": "$Default"
    },
    { 
      "name": "referenceTable",
      "type": "sql",
      "direction": "in",
      "commandText": "GetPlayerStats",
      "commandType":"storedProcedure",
      "parameters": "@player_id = {PartitionKey}",
      "connectionStringSetting": "SqlConnectionString"
    }
  ]
}

修改后的Python代码 #

import logging
import json
import azure.functions as func


def main(event: func.EventHubEvent, referenceTable: func.SqlRowList):
    logging.info('Python EventHub trigger processed an event: %s',
                 event.get_body().decode('utf-8'))
    logging.info(f"Event partition key: {event.partition_key}")
    
    rows = [json.loads(row.to_json()) for row in referenceTable]
    logging.info(f"Stored procedure result: {rows}")

方案二：保留批量处理，手动调用存储过程 #

如果需要保持批量接收事件的模式，不能依赖SQL输入绑定，需在代码中逐个获取事件的分区键，通过SQL客户端手动调用存储过程。

步骤1：安装依赖包 #

在函数项目根目录执行：

pip install pyodbc -t .

修改后的Python代码 #

from typing import List
import logging
import json
import azure.functions as func
import pyodbc
import os


def main(events: List[func.EventHubEvent]):
    # 从环境变量获取SQL连接字符串
    conn_str = os.environ["SqlConnectionString"]
    
    for event in events:
        event_body = event.get_body().decode('utf-8')
        partition_key = event.partition_key
        logging.info(f"Processed event: {event_body}, Partition key: {partition_key}")
        
        # 连接数据库并执行存储过程
        with pyodbc.connect(conn_str) as conn:
            with conn.cursor() as cursor:
                cursor.execute("{CALL GetPlayerStats(?)}", partition_key)
                rows = cursor.fetchall()
                
                # 处理结果
                result = []
                for row in rows:
                    result.append(dict(zip([column[0] for column in cursor.description], row)))
                logging.info(f"Stored procedure result: {result}")

注意事项 #

• 确保函数应用的环境变量中已配置SqlConnectionString
• 根据SQL Server版本调整pyodbc的连接参数（如需要添加Driver={ODBC Driver 17 for SQL Server}）

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