4

我正在尝试使用 Azure 数据工厂 (v2) 备份我的 Cosmos Db 存储。一般来说,它正在做它的工作,但我想让 Cosmos 集合中的每个文档对应于 blob 存储中的新 json 文件。

使用下一个复制参数,我可以将集合中的所有文档复制到 azure blob 存储中的 1 个文件中:

{
"name": "ForEach_mih",
"type": "ForEach",
"typeProperties": {
    "items": {
        "value": "@pipeline().parameters.cw_items",
        "type": "Expression"
    },
    "activities": [
        {
            "name": "Copy_mih",
            "type": "Copy",
            "policy": {
                "timeout": "7.00:00:00",
                "retry": 0,
                "retryIntervalInSeconds": 30,
                "secureOutput": false
            },
            "userProperties": [
                {
                    "name": "Source",
                    "value": "@{item().source.collectionName}"
                },
                {
                    "name": "Destination",
                    "value": "cosmos-backup-v2/@{item().destination.fileName}"
                }
            ],
            "typeProperties": {
                "source": {
                    "type": "DocumentDbCollectionSource",
                    "nestingSeparator": "."
                },
                "sink": {
                    "type": "BlobSink"
                },
                "enableStaging": false,
                "enableSkipIncompatibleRow": true,
                "redirectIncompatibleRowSettings": {
                    "linkedServiceName": {
                        "referenceName": "Clear_Test_BlobStorage",
                        "type": "LinkedServiceReference"
                    },
                    "path": "cosmos-backup-logs"
                },
                "cloudDataMovementUnits": 0
            },
            "inputs": [
                {
                    "referenceName": "SourceDataset_mih",
                    "type": "DatasetReference",
                    "parameters": {
                        "cw_collectionName": "@item().source.collectionName"
                    }
                }
            ],
            "outputs": [
                {
                    "referenceName": "DestinationDataset_mih",
                    "type": "DatasetReference",
                    "parameters": {
                        "cw_fileName": "@item().destination.fileName"
                    }
                }
            ]
        }
    ]
}
}

如何将每个 cosmos doc 复制到单独的文件并将其命名为 {PartitionId}-{docId}?

UPD

源集代码:

{
"name": "ClustersData",
"properties": {
    "linkedServiceName": {
        "referenceName": "Clear_Test_CosmosDb",
        "type": "LinkedServiceReference"
    },
    "type": "DocumentDbCollection",
    "typeProperties": {
        "collectionName": "directory-clusters"
    }
}
}

目的地集代码:

{
"name": "OutputClusters",
"properties": {
    "linkedServiceName": {
        "referenceName": "Clear_Test_BlobStorage",
        "type": "LinkedServiceReference"
    },
    "type": "AzureBlob",
    "typeProperties": {
        "format": {
            "type": "JsonFormat",
            "filePattern": "arrayOfObjects"
        },
        "fileName": "",
        "folderPath": "cosmos-backup-logs"
    }
}
}

管道代码:

{
"name": "copy-clsts",
"properties": {
    "activities": [
        {
            "name": "LookupClst",
            "type": "Lookup",
            "policy": {
                "timeout": "7.00:00:00",
                "retry": 0,
                "retryIntervalInSeconds": 30,
                "secureOutput": false
            },
            "typeProperties": {
                "source": {
                    "type": "DocumentDbCollectionSource",
                    "nestingSeparator": "."
                },
                "dataset": {
                    "referenceName": "ClustersData",
                    "type": "DatasetReference"
                },
                "firstRowOnly": false
            }
        },
        {
            "name": "ForEachClst",
            "type": "ForEach",
            "dependsOn": [
                {
                    "activity": "LookupClst",
                    "dependencyConditions": [
                        "Succeeded"
                    ]
                }
            ],
            "typeProperties": {
                "items": {
                    "value": "@activity('LookupClst').output.value",
                    "type": "Expression"
                },
                "batchCount": 8,
                "activities": [
                    {
                        "name": "CpyClst",
                        "type": "Copy",
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30,
                            "secureOutput": false
                        },
                        "typeProperties": {
                            "source": {
                                "type": "DocumentDbCollectionSource",
                                "query": "select @{item()}",
                                "nestingSeparator": "."
                            },
                            "sink": {
                                "type": "BlobSink"
                            },
                            "enableStaging": false,
                            "enableSkipIncompatibleRow": true,
                            "cloudDataMovementUnits": 0
                        },
                        "inputs": [
                            {
                                "referenceName": "ClustersData",
                                "type": "DatasetReference"
                            }
                        ],
                        "outputs": [
                            {
                                "referenceName": "OutputClusters",
                                "type": "DatasetReference"
                            }
                        ]
                    }
                ]
            }
        }
    ]
}
}

输入集合中的文档示例(格式相同):

{
   "$type": "Entities.ADCluster",
    "DisplayName": "TESTNetBIOS",
    "OrgId": "9b679d2a-42c5-4c9a-a2e2-3ce63c1c3506",
    "ClusterId": "ab2a242d-f1a5-62ed-b420-31b52e958586",
    "AllowLdapLifeCycleSynchronization": true,
    "DirectoryServers": [
        {
            "$type": "Entities.DirectoryServer",
            "AddressId": "e6a8edbb-ad56-4135-94af-fab50b774256",
            "Port": 389,
            "Host": "192.168.342.234"
        }
    ],
    "DomainNames": [
        "TESTNetBIOS"
    ],
    "BaseDn": null,
    "UseSsl": false,
    "RepositoryType": 1,
    "DirectoryCustomizations": null,
    "_etag": "\"140046f2-0000-0000-0000-5ac63a180000\"",
    "LastUpdateTime": "2018-04-05T15:00:40.243Z",
    "id": "ab2a242d-f1a5-62ed-b420-31b52e958586",
    "PartitionKey": "directory-clusters-9b679d2a-42c5-4c9a-a2e2-3ce63c1c3506",
    "_rid": "kpvxLAs6gkmsCQAAAAAAAA==",
    "_self": "dbs/kvpxAA==/colls/kpvxLAs6gkk=/docs/kvpxALs6kgmsCQAAAAAAAA==/",
    "_attachments": "attachments/",
    "_ts": 1522940440
}
4

4 回答 4

2

您是否考虑过使用 Azure Functions 以不同的方式实现这一点?ADF 旨在将数据从一个地方批量移动到另一个地方,并且每个集合只生成一个文件。

您可以考虑在集合中添加/更新文档时触发 Azure 函数,并让 Azure 函数将文档输出到 blob 存储。这应该可以很好地扩展,并且相对容易实现。

于 2018-06-18T10:14:34.820 回答
2

由于您的 cosmosdb 具有数组,并且 ADF 不支持 cosmos db 的序列化数组,因此这是我可以提供的解决方法。

首先,使用 export json as-is 将所有文档导出到 json 文件(到 blob 或 adls 或文件系统,任何文件存储)。我想你已经知道该怎么做了。这样每个集合都会有一个json文件。

其次,处理每个json文件,将文件中的每一行精确到一个文件。

我只为第 2 步提供管道。您可以使用执行管道活动来链接第 1 步和第 2 步。您甚至可以使用 foreach 活动处理第 2 步中的所有集合。

管道json

{
"name": "pipeline27",
"properties": {
    "activities": [
        {
            "name": "Lookup1",
            "type": "Lookup",
            "policy": {
                "timeout": "7.00:00:00",
                "retry": 0,
                "retryIntervalInSeconds": 30,
                "secureOutput": false
            },
            "typeProperties": {
                "source": {
                    "type": "BlobSource",
                    "recursive": true
                },
                "dataset": {
                    "referenceName": "AzureBlob7",
                    "type": "DatasetReference"
                },
                "firstRowOnly": false
            }
        },
        {
            "name": "ForEach1",
            "type": "ForEach",
            "dependsOn": [
                {
                    "activity": "Lookup1",
                    "dependencyConditions": [
                        "Succeeded"
                    ]
                }
            ],
            "typeProperties": {
                "items": {
                    "value": "@activity('Lookup1').output.value",
                    "type": "Expression"
                },
                "activities": [
                    {
                        "name": "Copy1",
                        "type": "Copy",
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30,
                            "secureOutput": false
                        },
                        "typeProperties": {
                            "source": {
                                "type": "DocumentDbCollectionSource",
                                "query": {
                                    "value": "select @{item()}",
                                    "type": "Expression"
                                },
                                "nestingSeparator": "."
                            },
                            "sink": {
                                "type": "BlobSink"
                            },
                            "enableStaging": false,
                            "cloudDataMovementUnits": 0
                        },
                        "inputs": [
                            {
                                "referenceName": "DocumentDbCollection1",
                                "type": "DatasetReference"
                            }
                        ],
                        "outputs": [
                            {
                                "referenceName": "AzureBlob6",
                                "type": "DatasetReference",
                                "parameters": {
                                    "id": {
                                        "value": "@item().id",
                                        "type": "Expression"
                                    },
                                    "PartitionKey": {
                                        "value": "@item().PartitionKey",
                                        "type": "Expression"
                                    }
                                }
                            }
                        ]
                    }
                ]
            }
        }
    ]
},
"type": "Microsoft.DataFactory/factories/pipelines"

}

用于查找的数据集 json

   {
"name": "AzureBlob7",
"properties": {
    "linkedServiceName": {
        "referenceName": "bloblinkedservice",
        "type": "LinkedServiceReference"
    },
    "type": "AzureBlob",
    "typeProperties": {
        "format": {
            "type": "JsonFormat",
            "filePattern": "arrayOfObjects"
        },
        "fileName": "cosmos.json",
        "folderPath": "aaa"
    }
},
"type": "Microsoft.DataFactory/factories/datasets"

}

用于复制的源数据集。实际上,这个数据集没有用。只想用它来托管查询(选择@{item()}

{
"name": "DocumentDbCollection1",
"properties": {
    "linkedServiceName": {
        "referenceName": "CosmosDB-r8c",
        "type": "LinkedServiceReference"
    },
    "type": "DocumentDbCollection",
    "typeProperties": {
        "collectionName": "test"
    }
},
"type": "Microsoft.DataFactory/factories/datasets"

}

目标数据集。使用两个参数,它还解决了您的文件名请求。

{
"name": "AzureBlob6",
"properties": {
    "linkedServiceName": {
        "referenceName": "AzureStorage-eastus",
        "type": "LinkedServiceReference"
    },
    "parameters": {
        "id": {
            "type": "String"
        },
        "PartitionKey": {
            "type": "String"
        }
    },
    "type": "AzureBlob",
    "typeProperties": {
        "format": {
            "type": "JsonFormat",
            "filePattern": "setOfObjects"
        },
        "fileName": {
            "value": "@{dataset().PartitionKey}-@{dataset().id}.json",
            "type": "Expression"
        },
        "folderPath": "aaacosmos"
    }
},
"type": "Microsoft.DataFactory/factories/datasets"

}

另请注意查找活动的限制:查找支持以下数据源。Lookup 活动可以返回的最大行数为 5000,最大为 2MB。目前,超时前查找活动的最长持续时间为一小时。

于 2018-06-17T11:51:29.480 回答
0

仅以一个集合为例。 在此处输入图像描述

在 foreach 内部: 在此处输入图像描述

并且您的查找和复制活动源数据集引用相同的 cosmosdb 数据集。

如果要复制 5 个集合,可以将此管道放入执行活动中。并且执行活动的主管道有一个 foreach 活动。

于 2018-06-15T14:00:52.800 回答
0

我也为此有些挣扎,尤其是绕过 Lookup 活动的大小限制,因为我们有很多数据要迁移。我最终创建了一个带有时间戳列表的 JSON 文件,用于查询 Cosmos 数据,然后为每个时间戳获取该范围内的文档 ID,然后为每个时间戳获取完整的文档数据并将其保存到路径如PartitionKey/DocumentID。这是我创建的管道:

LookupTimestamps - 遍历times.json文件中的每个时间戳范围,对于每个时间戳,执行 ExportFromCosmos 管道

{
    "name": "LookupTimestamps",
    "properties": {
        "activities": [
            {
                "name": "LookupTimestamps",
                "type": "Lookup",
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "typeProperties": {
                    "source": {
                        "type": "BlobSource",
                        "recursive": false
                    },
                    "dataset": {
                        "referenceName": "BlobStorageTimestamps",
                        "type": "DatasetReference"
                    },
                    "firstRowOnly": false
                }
            },
            {
                "name": "ForEachTimestamp",
                "type": "ForEach",
                "dependsOn": [
                    {
                        "activity": "LookupTimestamps",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "typeProperties": {
                    "items": {
                        "value": "@activity('LookupTimestamps').output.value",
                        "type": "Expression"
                    },
                    "isSequential": false,
                    "activities": [
                        {
                            "name": "Execute Pipeline1",
                            "type": "ExecutePipeline",
                            "typeProperties": {
                                "pipeline": {
                                    "referenceName": "ExportFromCosmos",
                                    "type": "PipelineReference"
                                },
                                "waitOnCompletion": true,
                                "parameters": {
                                    "From": {
                                        "value": "@{item().From}",
                                        "type": "Expression"
                                    },
                                    "To": {
                                        "value": "@{item().To}",
                                        "type": "Expression"
                                    }
                                }
                            }
                        }
                    ]
                }
            }
        ]
    },
    "type": "Microsoft.DataFactory/factories/pipelines"
}

ExportFromCosmos - 从上述管道执行的嵌套管道。这是为了解决您不能嵌套 ForEach 活动的事实。

{
    "name": "ExportFromCosmos",
    "properties": {
        "activities": [
            {
                "name": "LookupDocuments",
                "type": "Lookup",
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "typeProperties": {
                    "source": {
                        "type": "DocumentDbCollectionSource",
                        "query": {
                            "value": "select c.id, c.partitionKey from c where c._ts >= @{pipeline().parameters.from} and c._ts <= @{pipeline().parameters.to} order by c._ts desc",
                            "type": "Expression"
                        },
                        "nestingSeparator": "."
                    },
                    "dataset": {
                        "referenceName": "CosmosDb",
                        "type": "DatasetReference"
                    },
                    "firstRowOnly": false
                }
            },
            {
                "name": "ForEachDocument",
                "type": "ForEach",
                "dependsOn": [
                    {
                        "activity": "LookupDocuments",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "typeProperties": {
                    "items": {
                        "value": "@activity('LookupDocuments').output.value",
                        "type": "Expression"
                    },
                    "activities": [
                        {
                            "name": "Copy1",
                            "type": "Copy",
                            "policy": {
                                "timeout": "7.00:00:00",
                                "retry": 0,
                                "retryIntervalInSeconds": 30,
                                "secureOutput": false,
                                "secureInput": false
                            },
                            "typeProperties": {
                                "source": {
                                    "type": "DocumentDbCollectionSource",
                                    "query": {
                                        "value": "select * from c where c.id = \"@{item().id}\" and c.partitionKey = \"@{item().partitionKey}\"",
                                        "type": "Expression"
                                    },
                                    "nestingSeparator": "."
                                },
                                "sink": {
                                    "type": "BlobSink"
                                },
                                "enableStaging": false
                            },
                            "inputs": [
                                {
                                    "referenceName": "CosmosDb",
                                    "type": "DatasetReference"
                                }
                            ],
                            "outputs": [
                                {
                                    "referenceName": "BlobStorageDocuments",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "id": {
                                            "value": "@item().id",
                                            "type": "Expression"
                                        },
                                        "partitionKey": {
                                            "value": "@item().partitionKey",
                                            "type": "Expression"
                                        }
                                    }
                                }
                            ]
                        }
                    ]
                }
            }
        ],
        "parameters": {
            "from": {
                "type": "int"
            },
            "to": {
                "type": "int"
            }
        }
    }
}

BlobStorageTimestamps -times.json文件的数据集

{
    "name": "BlobStorageTimestamps",
    "properties": {
        "linkedServiceName": {
            "referenceName": "AzureBlobStorage1",
            "type": "LinkedServiceReference"
        },
        "type": "AzureBlob",
        "typeProperties": {
            "format": {
                "type": "JsonFormat",
                "filePattern": "arrayOfObjects"
            },
            "fileName": "times.json",
            "folderPath": "mycollection"
        }
    },
    "type": "Microsoft.DataFactory/factories/datasets"
}

BlobStorageDocuments - 用于保存文档的数据集

{
    "name": "BlobStorageDocuments",
    "properties": {
        "linkedServiceName": {
            "referenceName": "AzureBlobStorage1",
            "type": "LinkedServiceReference"
        },
        "parameters": {
            "id": {
                "type": "string"
            },
            "partitionKey": {
                "type": "string"
            }
        },
        "type": "AzureBlob",
        "typeProperties": {
            "format": {
                "type": "JsonFormat",
                "filePattern": "arrayOfObjects"
            },
            "fileName": {
                "value": "@{dataset().partitionKey}/@{dataset().id}.json",
                "type": "Expression"
            },
            "folderPath": "mycollection"
        }
    },
    "type": "Microsoft.DataFactory/factories/datasets"
}

times.json 文件只是一个纪元时间列表,如下所示:

[{
    "From": 1556150400,
    "To": 1556236799
},
{
    "From": 1556236800,
    "To": 1556323199
}]
于 2019-04-30T14:14:28.167 回答