如何使用Azure Data Factory活动实现Service Bus触发的Python处理逻辑

用Azure Data Factory实现对应处理管道的步骤 #

你原来的9个处理步骤可以直接对应到ADF的不同活动实现：

• 步骤1（接收消息转JSON）：配置Service Bus Topic事件触发器启动管道，接收到消息后用ADF内置表达式@replace()先把消息体里的分号替换为逗号，再用@json()表达式将消息内容解析为JSON格式。
• 步骤2（合法性校验）：使用If Condition活动，判断条件配置为@and(equals(activity('解析JSON').output.error, json('{}')), not(equals(activity('解析JSON').output.MSG_TYPE_TAG, '')))，条件不满足的分支直接结束管道，可可选添加日志记录。
• 步骤3（二次条件校验）：承接上一步校验通过的分支，再添加一个If Condition活动，判断条件为@and(equals(int(activity('解析JSON').output.MSG_TYPE_TAG), 50), equals(activity('解析JSON').output.GET_RIGIDSENSE_SENSOR_ACCELDATA_LOG_TAG.logical_name, 'BLOCK'))，不满足条件同样直接结束管道。
• 步骤5（生成指定数组）：校验通过后用Set Variable活动，将JSON中GET_RIGIDSENSE_SENSOR_ACCELDATA_LOG_TAG.acceleration_array字段的取值存入数组类型的管道变量。
• 步骤6-8（特征提取、归一化、分类打标）：这三步涉及numpy、pandas计算以及预训练的pickle模型调用，ADF原生表达式无法直接实现，可通过Azure Function活动或者Databricks Notebook活动实现：把你现有代码中对应的逻辑抽出来封装为Function/Notebook，输入传步骤5得到的加速度数组，输出返回分类标签和置信概率即可。
• 步骤9（写入数据库）：使用ADF内置的Azure PostgreSQL连接器对应的Copy活动，把处理后的所有需要的字段映射到dci_output_lable表的对应列，自动完成写入操作，不需要手动编写数据库连接和提交代码。

其他场景建议 #

• 如果你当前的消息吞吐量不大、端到端处理时延要求在秒级，现有Azure Function的方案已经可以满足需求，不需要特意迁移到ADF：ADF更适合有批量处理、多数据源调度、复杂ETL流转的场景，单条流式消息处理用Function的成本和时延表现都更好。
• 不管用哪种方案，预训练的scaler.sav和ExtraTreesClassifier.sav模型文件建议存到Azure Blob存储中，计算环节直接挂载Blob读取，不要把模型打包到代码包中，方便后续模型迭代更新。
• 建议配置Service Bus死信队列：处理失败、校验不通过的消息都转入死信队列留存，方便后续排查问题和回溯数据。
• 数据库账号密码等敏感配置统一存到Azure Key Vault中，Function和ADF都直接从Key Vault读取配置，不要硬编码在代码或者配置文件中，提升安全性。

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