ADF Snowflake源复制活动动态查询报错：'iem'函数未定义

我有一个ADF复制活动（源为Snowflake，目标为Azure存储），通过变量传入schema和表名并使用动态查询。执行时出现错误：

'The template function 'iem' is not defined or not valid'

已知信息：

• Schema和表名必须添加双引号，否则Snowflake实例会拒绝请求；硬编码表名的查询（如SELECT * FROM "schema"."table"）可正常运行。
• 即使改为无参数的简单查询仍报错，但数据预览功能可正常工作。

以下是For Each循环的JSON配置：

{"name":"ForEachINCTable","type":"ForEach","dependsOn":[{"activity":"LookupINCTable","dependencyConditions":["Succeeded"]}],"userProperties":[],"typeProperties":{"items":{"value":"@activity('LookupINCTable').output.value","type":"Expression"},"batchCount":10,"activities":[{"name":"WriteCopyLogInc","type":"SqlServerStoredProcedure","dependsOn":[{"activity":"aCopyINC","dependencyConditions":["Completed"]}],"policy":{"timeout":"7.00:00:00","retry":1,"retryIntervalInSeconds":30,"secureOutput":false,"secureInput":false},"userProperties":[],"typeProperties":{"storedProcedureName":"Log.InsertADFActivityLog","storedProcedureParameters":{"01ID":{"value":{"value":"@activity('aCopyINC').ActivityRunId","type":"Expression"},"type":"String"},"02ProcessVariablesJSON":{"value":{"value":"@{item()}","type":"Expression"},"type":"String"},"03ActivityOutputJSON":{"value":{"value":"@{activity('aCopyINC')}","type":"Expression"},"type":"String"},"04PipelineVariablesJSON":{"value":{"value":"@{pipeline()}","type":"Expression"},"type":"String"}},"linkedServiceName":{"referenceName":"dnaMetadataLS","type":"LinkedServiceReference"}}},{"name":"AddFileDescriptionINC","type":"SqlServerStoredProcedure","dependsOn":[{"activity":"aCopyINC","dependencyConditions":["Succeeded"]}],"policy":{"timeout":"7.00:00:00","retry":1,"retryIntervalInSeconds":30,"secureOutput":false,"secureInput":false},"userProperties":[],"typeProperties":{"storedProcedureName":"[Metadata].InsertFileDescription","storedProcedureParameters":{"01TTID":{"value":{"value":"@item().SourceEntityId","type":"Expression"},"type":"Int64"},"02ADFRunId":{"value":{"value":"@pipeline().TriggerId","type":"Expression"},"type":"Guid"},"03Container":{"value":{"value":"@item().TargetSubSystem","type":"Expression"},"type":"String"},"04Directory":{"value":{"value":"@{item().TargetDirectory}","type":"Expression"},"type":"String"},"05File":{"value":{"value":"@{item().TargetFile}","type":"Expression"},"type":"String"},"06CreatedDate":{"value":{"value":"@formatDateTime(utcnow(),'yyyy-MM-dd HH:mm:ss')","type":"Expression"},"type":"DateTime"},"07LoadType":{"value":{"value":"@item().LoadType","type":"Expression"},"type":"String"},"08IncrementalFrom":{"value":{"value":"@item().IncrementalFrom","type":"Expression"},"type":"String"},"09IncrementalTo":{"value":{"value":"@item().IncrementalTo","type":"Expression"},"type":"String"},"10LoadStatusID":{"value":"2","type":"Int32"}},"linkedServiceName":{"referenceName":"dnaMetadataLS","type":"LinkedServiceReference"}}},{"name":"aCopyINC","type":"Copy","dependsOn":[{"activity":"Script Get Trigger Time","dependencyConditions":["Succeeded"]}],"policy":{"timeout":"7.00:00:00","retry":2,"retryIntervalInSeconds":30,"secureOutput":false,"secureInput":false},"userProperties":[{"name":"SourceSystem","value":"@{item().ConnectionID}"},{"name":"Table","value":"@{item().SourceEntityName}"},{"name":"TTID","value":"@{item().SourceEntityId}"}],"typeProperties":{"source":{"type":"SnowflakeV2Source","query":{"value":"@concat('SELECT * FROM \"', item().SourceSchemaName, '\".\"', item().SourceEntityName, '\" WHERE ', item().IncrementalColumn, ' >= ''', formatDateTime(replace(item().IncrementalFrom, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), ''' AND ', item().IncrementalColumn, ' < ''', formatDateTime(replace(item().IncrementalTo, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), '''')","type":"Expression"},"exportSettings":{"type":"SnowflakeExportCopyCommand"}},"sink":{"type":"DelimitedTextSink","storeSettings":{"type":"AzureBlobStorageWriteSettings"},"formatSettings":{"type":"DelimitedTextWriteSettings","quoteAllText":true,"fileExtension":".txt"}},"enableStaging":false},"inputs":[{"referenceName":"DSRedzone","type":"DatasetReference","parameters":{"tableName":{"value":"@item().SourceEntityName","type":"Expression"},"schemaName":{"value":"@item().SourceSchemaName","type":"Expression"}}}],"outputs":[{"referenceName":"DSBlobStorage_SAS_csv","type":"DatasetReference","parameters":{"Container":{"value":"@item().TargetSubSystem","type":"Expression"},"Directory":{"value":"@{item().TargetDirectory}","type":"Expression"},"File":{"value":"@{item().TargetFile}","type":"Expression"},"Separator":",","HeaderRow":"true","Escape":"\"","Quote":"\""}}]}},{"name":"Update Incremental Value","type":"Script","dependsOn":[{"activity":"aCopyINC","dependencyConditions":["Succeeded"]}],"policy":{"timeout":"0.12:00:00","retry":0,"retryIntervalInSeconds":30,"secureOutput":false,"secureInput":false},"userProperties":[],"linkedServiceName":{"referenceName":"dnaMetadataLS","type":"LinkedServiceReference"},"typeProperties":{"scripts":[{"type":"NonQuery","text":{"value":"UPDATE [Metadata].[IncrementalState]\nSET [LastRunIncrementalFrom] = CONVERT(DATETIME2(0),'@{item().IncrementalFrom}',120),\n    [LastRunIncrementalTo] = CONVERT(DATETIME2(0), SYSDATETIME(),120)\nWHERE SourceEntityID = @{item().SourceEntityId} ;\n\nUPDATE [Metadata].[IncrementalState]\nSET [IncrementalFrom] = CONVERT(DATETIME2(0), SYSDATETIME(), 120),\n    [IncrementalTo] = '9999-12-31 00:00:00'\nWHERE SourceEntityID = @{item().SourceEntityId} ;","type":"Expression"}}],"scriptBlockExecutionTimeout":"02:00:00"}},{"name":"Script Get Trigger Time","type":"Script","dependsOn":[],"policy":{"timeout":"0.12:00:00","retry":0,"retryIntervalInSeconds":30,"secureOutput":false,"secureInput":false},"userProperties":[],"linkedServiceName":{"referenceName":"dnaMetadataLS","type":"LinkedServiceReference"},"typeProperties":{"scripts":[{"type":"Query","text":{"value":"DECLARE @@TriggerTime AS Datetime2(0);\n\nSELECT @@TriggerTime = TriggerTime \nFROM [Metadata].[PipelineExecutionLastSuccessfull]\nWHERE pipeline='@{pipeline().Pipeline}';\n\n","type":"Expression"}}],"scriptBlockExecutionTimeout":"02:00:00"}}]}]}

错误根源分析 #

错误提示中的iem函数并非ADF内置函数，大概率是表达式拼写错误：

1. 检查所有@item()的使用场景，是否存在误写为@iem()的情况——虽然提供的JSON里显示都是item()，但可能在可视化编辑时出现输入错误，或JSON复制时的字符损坏。
2. 另一种可能是ADF在解析嵌套表达式时的语法冲突：比如存储过程参数或脚本中的@{item()}表达式，是否存在未正确闭合的引号或语法错误。

具体修复步骤 #

1. 检查所有表达式中的item()拼写 #

遍历ForEach循环内所有使用item()的位置，确认没有拼写错误：

• 复制活动的用户属性
• 动态查询的concat表达式
• 存储过程参数
• Script活动的SQL文本

2. 优化动态查询的引号转义（避免出错） #

当前动态查询的引号转义可以简化，减少语法错误概率：
原表达式：

@concat('SELECT * FROM \"', item().SourceSchemaName, '\".\"', item().SourceEntityName, '\" WHERE ', item().IncrementalColumn, ' >= ''', formatDateTime(replace(item().IncrementalFrom, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), ''' AND ', item().IncrementalColumn, ' < ''', formatDateTime(replace(item().IncrementalTo, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), '''')

修改为用单引号包裹Snowflake的双引号字符串（ADF中用两个单引号转义）：

@concat('SELECT * FROM ''"', item().SourceSchemaName, '"."', item().SourceEntityName, '"'' WHERE ', item().IncrementalColumn, ' >= ''', formatDateTime(replace(item().IncrementalFrom, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), ''' AND ', item().IncrementalColumn, ' < ''', formatDateTime(replace(item().IncrementalTo, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), '''')

生成的SQL会正确包含Snowflake要求的双引号。

3. 修正Script活动的表名拼写 #

Script Get Trigger Time活动的SQL中，表名PipelineExecutionLastSuccessfull存在拼写错误（多了一个'l'），应改为PipelineExecutionLastSuccessful：

DECLARE @@TriggerTime AS Datetime2(0);
SELECT @@TriggerTime = TriggerTime 
FROM [Metadata].[PipelineExecutionLastSuccessful]
WHERE pipeline='@{pipeline().Pipeline}';

4. 重置ForEach循环的表达式缓存 #

ADF有时会出现表达式缓存问题，导致解析错误：

1. 进入ForEach活动的设置页面，重新输入@activity('LookupINCTable').output.value表达式，保存后重新发布管道。
2. 清空ADF的运行历史，重新触发管道测试。

验证方法 #

1. 测试单个表的复制：暂时修改ForEach的items为固定数组，比如[{"SourceSchemaName":"test_schema","SourceEntityName":"test_table",...}]，确认动态查询可以正常生成并执行。
2. 使用ADF的测试表达式功能，在动态查询的表达式编辑器中测试，查看生成的SQL是否符合预期。

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