如何使用Azure Data Factory将CosmosDb文档复制到Blob存储（单个json文件中的每个文档）

问题描述 投票：1回答：4

我正在尝试使用Azure Data Factory（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"
                            }
                        ]
                    }
                ]
            }
        }
    ]
}
}

输入集合中的doc示例（所有格式相同）：

{
   "$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
}
azure azure-cosmosdb azure-data-factory-2
4个回答
2
投票

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

首先，使用export json as（是blob或adls或文件系统，任何文件存储）将所有文档导出到json文件。我想你已经知道怎么做了。这样，每个集合都将有一个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活动的限制：查找支持以下数据源。 Lookup活动可以返回的最大行数为5000，最大为2MB。目前，超时前Lookup活动的最长持续时间为一小时。

2
投票

您是否考虑过使用Azure功能以不同的方式实现此功能？ ADF旨在将批量数据从一个位置移动到另一个位置，并且每个集合仅生成一个文件。

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

0
投票

以一个集合为例。 enter image description here

在foreach内部：enter image description here

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

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

0
投票

我也有点挣扎，特别是绕过Lookup活动的大小限制，因为我们有很多要迁移的数据。我最终创建了一个带有时间戳列表的JSON文件来查询Cosmos数据，然后为每个时间戳获取该范围内的文档ID，然后为每个文档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
}]
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.