如何在Azure Stream Analytics Job中将Blob配置数据传入UDF函数

以下两种方案都可以满足需求，优先推荐第一种：

方案1：固定键关联参考数据（最稳定，性能无损耗） #

你担心的JOIN效率问题仅在参考数据量大、关联键离散时才会出现，单条配置的关联开销可以忽略不计：

• 首先修改Blob存储中的JSON配置，外层增加一个固定值的关联字段，比如"JoinKey": "static"，确保参考数据加载后只有1行记录
• 查询时给流数据也拼接相同的固定关联键，再做关联即可，参考数据只会在作业启动、或你配置的刷新间隔时重新加载一次，运行时关联为纯内存操作，完全不影响流式处理性能
  示例查询代码如下：

-- 加载Blob存储的配置参考数据
ConfigData AS (
    SELECT * FROM BlobConfigInput
),

-- 流数据与配置关联
DataWithConfig AS (
    SELECT
        d.*,
        c.* AS ConfigFromBlob
    FROM (SELECT *, "static" AS JoinKey FROM DataFromPreviousStep) d
    JOIN (SELECT *, "static" AS JoinKey FROM ConfigData) c
    ON d.JoinKey = c.JoinKey
),

-- 计算时直接传入配置即可
SomeData AS (  
    SELECT
        Data, 
        UDF.PerformSomeCalculations(Data.Cars, Data.Trains, ConfigFromBlob) AS ProcessedData
    FROM DataWithConfig
)

该方案还支持配置自动热更新，只要在参考数据输入配置中设置对应的刷新频率，ASA会自动拉取新版本的配置，无需重启作业。

方案2：JS UDF内部缓存配置 #

如果你不想在查询逻辑中加关联步骤，可以在UDF中实现配置的拉取和缓存：

• 提前给配置Blob生成公共读权限，或者带有效期的SAS访问链接
• 在JS UDF中使用全局变量缓存配置，仅在缓存失效时重新拉取，示例UDF代码如下：

// 全局缓存变量，仅UDF初始化/刷新时赋值
let cachedConfig = null;
let lastFetchTime = 0;
const CONFIG_CACHE_TTL = 3600 * 1000; // 配置缓存1小时，可按需调整

function PerformSomeCalculations(cars, trains) {
    const currentTime = Date.now();
    // 缓存为空或过期时拉取新配置
    if (!cachedConfig || currentTime - lastFetchTime > CONFIG_CACHE_TTL) {
        const configBlobUrl = "https://<你的存储账户名>.blob.core.windows.net/<配置容器名>/config.json";
        const response = require('http').get(configBlobUrl);
        cachedConfig = JSON.parse(response.readAll());
        lastFetchTime = currentTime;
    }
    // 此处使用cachedConfig完成后续计算逻辑
    // ...
}

使用该方案需要提前在ASA作业的配置中开启UDF的外部访问权限，同时确保ASA的网络策略允许访问对应的Blob存储。

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