如何使用Azure Data Factory将Cosmos DB文档逐个导出为以文档ID命名的JSON格式Blob文件？

Your approach of using Lookup + ForEach + Copy Data is exactly the right way to achieve this requirement. Let’s break down the configuration step by step, addressing your specific questions along the way:

1. Configure the Lookup Activity (Find) #

First, we need to retrieve all documents from your Cosmos DB collection:

• Use your existing Cosmos DB dataset as the source.
• In the Query field, enter SELECT * FROM c (adjust with filters if you only need a subset of documents).
• Under Settings, uncheck "First row only" to fetch all documents.
  Note: Lookup has a default row limit of 1000, which can be increased to 5000 in the settings. If you have more than 5000 documents, you’ll need to use a pagination approach (e.g., Copy Data to a temporary blob first, then iterate over that), but this works for most common scenarios.

2. Configure the ForEach Activity #

Next, we’ll iterate over each document from the Lookup:

• Set the Items expression to @activity('Find').output.value (not .id—this gives you the entire document object per iteration, which we need to both get the id and the document content).
• If you want to avoid throttling Cosmos DB or Blob Storage, set Sequential to Yes (default is parallel with 20 concurrent runs, which is fine for smaller datasets).

3. Copy Data Activity Inside ForEach #

This is where we handle copying each individual document to Blob Storage. You can absolutely use your existing datasets here—we just need to parameterize them dynamically:

Source (Cosmos DB) #

• Use your existing Cosmos dataset.
• In the Source tab of the Copy Data activity, set the Query to:

  @concat('SELECT * FROM c WHERE c.id = "', item().id, '"')
  

  This ensures we only fetch the specific document from the current iteration.
• Disable "Read Cosmos DB items in parallel" (unnecessary for single-document queries).

Sink (Blob Storage) #

Your existing Blob dataset needs a parameter to dynamically set the filename. Here’s how to set it up:

1. Edit your Blob dataset:
   • Add a new parameter (e.g., TargetFileName).
   • Update the File path field to use this parameter. For example, if your original path was my-container/static-file.json, change it to:

     @concat('my-container/', dataset().TargetFileName)
     

2. Back in the Copy Data activity’s Sink tab:
   • Pass the parameter value as @concat(item().id, '.json')—this generates the filename using the document’s id with a .json suffix.
   • Set Write behavior to "Overwrite" if you want to replace existing files with the same id (adjust as needed).

Key Clarifications for Your Questions #

• Can ForEach use Copy Data with existing datasets? Yes—by adding parameters to your Blob dataset and using dynamic expressions in the Copy Data activity, you can reuse your existing datasets without recreating them.
• Why not use @activity('Find').output.value.id for ForEach items? That expression would only pass the list of id values, not the full document. We need the entire document to both filter the Cosmos source and set the filename, so using @activity('Find').output.value (the full list of documents) is correct.

Once all steps are configured, running the pipeline will generate a separate .json file in Blob Storage for each Cosmos DB document, with the filename matching the document’s id.

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