【发布时间】：2023-02-17 05:52:55
【问题描述】：
以下是获取元数据活动的输出，其中包含子项的名称和类型值：
是否可以只获取名称值并将其存储在数组变量中而不使用任何迭代。
输出 = [csv1.csv、csv2.csv、csv3.csv、csv4.csv]
这是通过 Foreach 和附加变量实现的，我们不想使用迭代。
【问题讨论】：
标签： azure-data-factory azure-synapse
【发布时间】：2023-02-17 05:52:55
【问题描述】：
方法 1：
使用 for each 会更容易完成工作。但是，您可以通过以下方式使用字符串操作来获得所需的结果。
- 将获取元数据子项的输出作为字符串存储在变量中：
@string(activity('Get Metadata1').output.childItems)
- 现在使用以下动态内容将所有不必要的数据替换为空字符串
''：
@replace(replace(replace(replace(replace(replace(replace(replace(variables('tp'),'[{',''),'}]',''),'{',''),'}',''),'"type":"File"',''),'"',''),'name:',''),',,',',')
- 现在，忽略最后一个逗号并使用
,作为分隔符拆分上面的字符串。
@split(substring(variables('ans'),0,sub(length(variables('ans')),1)),',')
方法 2：
假设您的源包含文件夹和文件的组合，并且您只需要数组中类型为 File 的对象的名称，那么您可以使用以下方法。这里不需要 for each，但是你将不得不使用复制数据和数据流。
- 使用包含如下数据的示例文件创建复制数据活动：
- 现在创建一个附加列
my_json，其值为以下动态内容：
@replace(string(activity('Get Metadata1').output.childItems),'"',pipeline().parameters.single_quote)
- 以下是我采用的接收器数据集配置：
- 在映射中，只需选择这个新创建的列并删除其余（演示）列。
- 执行此复制数据后，生成的文件将如下所示：
- 在数据流中，以上述文件为源，设置如下图所示：
- 读取的数据如下所示：
- 现在，使用聚合转换按类型列和
name列上的 collect() 进行分组。
- 结果如下所示：
- 现在，使用条件拆分来分离文件类型数据和文件夹类型数据，条件为
type == 'File'
- 现在将文件类型数据写入接收器缓存。数据看起来像这样：
- 回到管道中，使用以下动态内容获取所需的数组：
@activity('Data flow1').output.runStatus.output.sink1.value[0].array_of_types
管道 JSON 供参考：
{
"name": "pipeline3",
"properties": {
"activities": [
{
"name": "Get Metadata1",
"type": "GetMetadata",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"dataset": {
"referenceName": "source1",
"type": "DatasetReference"
},
"fieldList": [
"childItems"
],
"storeSettings": {
"type": "AzureBlobFSReadSettings",
"recursive": true,
"enablePartitionDiscovery": false
},
"formatSettings": {
"type": "DelimitedTextReadSettings"
}
}
},
{
"name": "Copy data1",
"type": "Copy",
"dependsOn": [
{
"activity": "Get Metadata1",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "DelimitedTextSource",
"additionalColumns": [
{
"name": "my_json",
"value": {
"value": "@replace(string(activity('Get Metadata1').output.childItems),'"',pipeline().parameters.single_quote)",
"type": "Expression"
}
}
],
"storeSettings": {
"type": "AzureBlobFSReadSettings",
"recursive": true,
"enablePartitionDiscovery": false
},
"formatSettings": {
"type": "DelimitedTextReadSettings"
}
},
"sink": {
"type": "DelimitedTextSink",
"storeSettings": {
"type": "AzureBlobFSWriteSettings"
},
"formatSettings": {
"type": "DelimitedTextWriteSettings",
"quoteAllText": true,
"fileExtension": ".txt"
}
},
"enableStaging": false,
"translator": {
"type": "TabularTranslator",
"mappings": [
{
"source": {
"name": "my_json",
"type": "String"
},
"sink": {
"type": "String",
"physicalType": "String",
"ordinal": 1
}
}
],
"typeConversion": true,
"typeConversionSettings": {
"allowDataTruncation": true,
"treatBooleanAsNumber": false
}
}
},
"inputs": [
{
"referenceName": "csv1",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "sink1",
"type": "DatasetReference"
}
]
},
{
"name": "Data flow1",
"type": "ExecuteDataFlow",
"dependsOn": [
{
"activity": "Copy data1",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"dataflow": {
"referenceName": "dataflow2",
"type": "DataFlowReference"
},
"compute": {
"coreCount": 8,
"computeType": "General"
},
"traceLevel": "None"
}
},
{
"name": "Set variable2",
"type": "SetVariable",
"dependsOn": [
{
"activity": "Data flow1",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"variableName": "req",
"value": {
"value": "@activity('Data flow1').output.runStatus.output.sink1.value[0].array_of_types",
"type": "Expression"
}
}
}
],
"parameters": {
"single_quote": {
"type": "string",
"defaultValue": "'"
}
},
"variables": {
"req": {
"type": "Array"
},
"tp": {
"type": "String"
},
"ans": {
"type": "String"
},
"req_array": {
"type": "Array"
}
},
"annotations": [],
"lastPublishTime": "2023-02-03T06:09:07Z"
},
"type": "Microsoft.DataFactory/factories/pipelines"
}
数据流 JSON 供参考：
{
"name": "dataflow2",
"properties": {
"type": "MappingDataFlow",
"typeProperties": {
"sources": [
{
"dataset": {
"referenceName": "Json3",
"type": "DatasetReference"
},
"name": "source1"
}
],
"sinks": [
{
"name": "sink1"
}
],
"transformations": [
{
"name": "aggregate1"
},
{
"name": "split1"
}
],
"scriptLines": [
"source(output(",
" name as string,",
" type as string",
" ),",
" allowSchemaDrift: true,",
" validateSchema: false,",
" ignoreNoFilesFound: false,",
" documentForm: 'arrayOfDocuments',",
" singleQuoted: true) ~> source1",
"source1 aggregate(groupBy(type),",
" array_of_types = collect(name)) ~> aggregate1",
"aggregate1 split(type == 'File',",
" disjoint: false) ~> split1@(fileType, folderType)",
"split1@fileType sink(validateSchema: false,",
" skipDuplicateMapInputs: true,",
" skipDuplicateMapOutputs: true,",
" store: 'cache',",
" format: 'inline',",
" output: true,",
" saveOrder: 1) ~> sink1"
]
}
}
}
【讨论】：