Queue-Based Bulk Data Import

Overview

Large CSV imports often exceed cloud-flow timeout budgets when written one row at a time directly into Dataverse. This flow solves that by fanning the import across two flows joined by an Azure Storage Queue: a producer flow reads the CSV from SharePoint, splits it into per-row JSON messages, and enqueues each one; a consumer flow drains the queue on its own schedule and writes each row into the target Dataverse table.

This page documents the producer side of the pattern. It is the flow that ships in the FlowLibs solution FlowLibsQueueBasedBulkDataImport.

Use Case

This pattern fits scenarios where a single, monolithic import flow would time out or be hard to retry — the Azure Queue gives you a durable buffer between the CSV source and the Dataverse sink, plus per-row retry granularity.

The flow is ideal for teams that:

• A nightly system export drops a .csv in a SharePoint document library and you need each row landed in a Dataverse table.
• A user-triggered bulk upload of 5,000+ records would otherwise time out a single flow.
• You need backpressure / throttling control between the source (SharePoint file drop) and the sink (Dataverse table) — Azure Queues give you a durable buffer between the two.
• You want to retry individual failed rows without re-processing the whole file.

Flow Architecture

Environment Variables

Connectors & Connections

Note — All connections are referenced as solution connection references; the flow is portable between environments as long as a connection is mapped at import time.

Customization Guide

Almost every realistic variant of this flow can be implemented by changing environment variable values. A few cases require small edits inside the flow definition — those are called out explicitly below.

Switching to an event-driven SharePoint trigger

Replace the manual trigger with OnNewFile (operationId OnNewFile on shared_sharepointonline) bound to the varLibraryPath folder, then change Get_CSV_File_Content to GetFileContent and bind its id parameter to triggerOutputs()?['body/Id']. Filter to .csv extensions with an If guard wrapping the parse pipeline using @endsWith(toLower(triggerOutputs()?['headers']?['x-ms-file-name']), '.csv').

Tuning row throughput

Default concurrency on Apply_To_Each_Row is 1 so the row counters stay accurate. If you don't need exact counts you can raise it to 20-50 for faster enqueueing, but Azure Queues throttles enqueue at the storage-account scope so very high concurrency is rarely worth the loss of counter accuracy.

Schema discipline of the queued message

Every message carries sourceFile, targetTable, headerRow, rowText, and enqueuedAt. The consumer flow can therefore (1) map fields by header position without re-reading SharePoint, (2) write to targetTable (lets one consumer service many imports), and (3) skip stale messages older than X hours via enqueuedAt.

Pairing with the consumer flow

Build a separate consumer flow with a Recurrence trigger (every 5 minutes), GetMessages_V2 (numofmessages: 32, visibilitytimeout: 60), Foreach over body('Get_Messages_From_Queue')?['QueueMessagesList']?['QueueMessage'] wrapped in coalesce(..., createArray()), parse the JSON message text, split rowText and headerRow on commas, build a Dataverse CreateRecord payload, and DeleteMessage_V2 only on success. Use the same flowlibs_AzureQueueStorageAccount and flowlibs_BulkImportQueueName env vars so the producer and consumer stay paired.

Multi-table fan-out

Promote flowlibs_BulkImportTargetTable to a column inside the CSV (or compute it from a folder convention like /BulkImports/Contacts/*.csv -> flowlibs_contacts), and stamp that value into the targetTable field of Compose_QueueMessage. The same producer can then drive imports into multiple tables.

Key Expressions

The flow is intentionally light on Power Fx / WDL gymnastics — the heaviest expressions are the branch-name concatenation and the approval outcome check. They are listed below in the order they appear in the flow.

@base64ToString(body('Get_CSV_File_Content')['$content'])

@split(replace(outputs('Compose_DecodedCsv'), decodeUriComponent('%0D'), ''), decodeUriComponent('%0A'))

@first(outputs('Compose_AllRows'))
@skip(outputs('Compose_AllRows'), 1)

@greater(length(trim(items('Apply_To_Each_Row'))), 0)

@string(outputs('Compose_QueueMessage'))

runAfter: { Try_Put_Message: ['Failed', 'TimedOut', 'Skipped'] }