如何通过ADF每10分钟调用Azure ML？解决ML pipeline读取旧数据问题

核心原因 #

你遇到的问题大概率是AML pipeline中Import Data模块的数据缓存机制导致的——默认情况下，该模块会缓存第一次读取的数据，后续运行不会主动重新拉取Blob中更新后的文件；另外如果你的Import Data固定死了文件路径/版本，也会导致无法读取新数据。这不是Datastore未更新的问题，而是数据读取环节的缓存或静态配置限制。

具体解决步骤 #

1. 快速修复：关闭Import Data的缓存 #

在AML studio打开你的ML pipeline，找到Import Data模块：

• 点击模块右上角的齿轮（设置）图标
• 取消勾选Enable caching选项
• 重新发布ML pipeline，再通过ADF调用验证

2. 替代Import Data：参数化路径+Python脚本读取数据 #

如果关闭缓存仍未解决，建议放弃Import Data模块，改用更灵活的参数化读取方案：

• 在AML pipeline中添加Execute Python Script模块，移除原Import Data
• 给AML pipeline添加字符串类型参数（比如input_file_path），用于接收ADF传递的Blob文件路径
• 在Python脚本中通过AML SDK读取Datastore的最新数据，示例代码：

import pandas as pd
from azureml.core import Run, Datastore

# 获取当前运行上下文
run = Run.get_context()
# 读取ADF传入的参数
input_file = run.get_parameter("input_file_path")
# 获取关联的Blob Datastore（替换为你的Datastore名称）
datastore = Datastore.get(run.experiment.workspace, datastore_name="your_blob_datastore")
# 下载最新文件到临时目录（覆盖旧文件）
datastore.download_files(target_path="./", prefix=input_file, overwrite=True)
# 读取CSV数据
df = pd.read_csv(input_file.split("/")[-1])

# 后续数据处理逻辑...

3. ADF向AML传递参数的配置 #

在ADF的Machine Learning Execute Pipeline活动中：

• 找到Pipeline parameters配置项
• 添加键值对：键为AML中定义的参数名（如input_file_path），值设为Blob中CSV的路径（可使用ADF动态内容，比如@variables('target_csv_path')，若为固定追加文件则直接写完整路径）

4. 跳过Blob中转：AML直接读取数据库数据 #

如果你的数据来源是数据库，可让AML直接连接数据库获取最新数据，省去ADF写Blob的环节：

• 在AML中创建连接目标数据库的Datastore（支持SQL Server、MySQL等主流数据源）
• 在Python脚本中直接通过Datastore连接数据库查询，示例代码：

import pandas as pd
from azureml.core import Run, Datastore
import pyodbc

run = Run.get_context()
# 获取数据库Datastore（替换为你的数据库Datastore名称）
db_datastore = Datastore.get(run.experiment.workspace, datastore_name="your_db_datastore")
# 获取数据库连接字符串
conn_str = db_datastore._get_connection_string()
# 连接并查询当日新增数据
conn = pyodbc.connect(conn_str)
query = "SELECT * FROM your_table WHERE record_date = CAST(GETDATE() AS DATE)"
df = pd.read_sql(query, conn)
conn.close()

# 后续数据处理逻辑...

5. Datastore有效性验证 #

若仍怀疑Datastore问题，可做以下验证：

• 在AML studio的Datastore页面点击Browse，确认能看到Blob容器中更新后的CSV文件
• 手动触发AML pipeline运行，查看日志中是否有读取文件的报错，或是否加载了正确的数据

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