Skip to main content

Queue ETL: Azure Queue Storage

Add an Azure Queue Storage connection string​

  • An Azure Queue Storage ETL task uses an Azure Queue Storage connection string to connect to its destination storage account. The connection string stores the storage account connection details and the authentication settings used to access Azure queues.

  • The ETL task does not store these connection details directly.
    Instead, its configuration references the connection string by name through the ConnectionStringName property.

  • An Azure Queue Storage connection string can be created in any of these ways:

  • The example below adds a per-database Azure Queue Storage connection string from the Client API.
    The connection string is stored in the current database record and can be used by tasks in that database.

Authentication methods​

An Azure Queue Storage connection string authenticates in one of three modes.
Set exactly one of these properties in AzureQueueStorageConnectionSettings:

  • ConnectionString:
    A single Azure Storage connection string.
    For http, it must include DefaultEndpointsProtocol, AccountName, AccountKey, and QueueEndpoint.
    For https, it must include DefaultEndpointsProtocol, AccountName, and the authentication details required by Azure.
  • EntraId:
    Microsoft Entra ID application credentials: StorageAccountName, TenantId, ClientId, and ClientSecret.
  • Passwordless:
    Passwordless machine authentication for self-hosted mode, using StorageAccountName.
    The machine account must have the Storage Account Queue Data Contributor role.

Example​

// Define an Azure Queue Storage connection string
// ===============================================
var azureQueueStorageConnectionString = new QueueConnectionString
{
    // The name used by the Azure Queue Storage ETL task configuration
    Name = "azure-queue-storage-connection-string-name",

    // Set the queue broker type to Azure Queue Storage
    BrokerType = QueueBrokerType.AzureQueueStorage,

    // Configure the Azure Queue Storage account connection
    AzureQueueStorageConnectionSettings = new AzureQueueStorageConnectionSettings
    {
        // A storage account connection string
        ConnectionString = "DefaultEndpointsProtocol=https;" +
                           "AccountName=<storage-account-name>;" +
                           "AccountKey=<storage-account-key>;" +
                           "EndpointSuffix=core.windows.net"
    }
};

// Deploy (send) the connection string to the server
// ==================================================
var putConnectionStringOp =
    new PutConnectionStringOperation<QueueConnectionString>(azureQueueStorageConnectionString);

PutConnectionStringResult connectionStringResult =
    store.Maintenance.Send(putConnectionStringOp);

Syntax​

public sealed class QueueConnectionString : ConnectionString
{
    // Set the broker type to QueueBrokerType.AzureQueueStorage
    // for an Azure Queue Storage connection string
    public QueueBrokerType BrokerType { get; set; }

    // Configure this when setting a connection string for Kafka
    public KafkaConnectionSettings KafkaConnectionSettings { get; set; }

    // Configure this when setting a connection string for RabbitMQ
    public RabbitMqConnectionSettings RabbitMqConnectionSettings { get; set; }

    // Configure this when setting a connection string for Azure Queue Storage
    public AzureQueueStorageConnectionSettings AzureQueueStorageConnectionSettings { get; set; }

    // Configure this when setting a connection string for Amazon SQS
    public AmazonSqsConnectionSettings AmazonSqsConnectionSettings { get; set; }

    // Configure this when setting a connection string for Azure Service Bus
    public AzureServiceBusConnectionSettings AzureServiceBusConnectionSettings { get; set; }
}

Add an Azure Queue Storage ETL task​

Example:

  • In this example, the Azure Queue Storage ETL Task will -
    • Extract source documents from the "Orders" collection in RavenDB.
    • Process each "Order" document using a defined script that creates a new orderData object.
    • Load the orderData object to the "OrdersQueue" in an Azure Queue Storage.
  • For more details about the script and the loadTo method, see the transformation script section below.
// Define a transformation script for the task: 
// ============================================
Transformation transformation = new Transformation
{
  // Define the input collections
  Collections = { "Orders" },
  ApplyToAllDocuments = false,
  
  // The transformation script
  Name = "scriptName",
  Script = @"// Create an orderData object
             // ==========================
             var orderData = {
                 Id: id(this),
                 OrderLinesCount: this.Lines.length,
                 TotalCost: 0
             };

             // Update the orderData's TotalCost field
             // ======================================
             for (var i = 0; i < this.Lines.length; i++) {
                 var line = this.Lines[i];
                 var cost = (line.Quantity * line.PricePerUnit) * ( 1 - line.Discount);
                 orderData.TotalCost += cost;
             }

             // Load the object to the 'OrdersQueue' in Azure
             // ============================================= 
             loadToOrdersQueue(orderData, {
                 Id: id(this),
                 Type: 'com.example.promotions',
                 Source: '/promotion-campaigns/summer-sale'
             });"
};

// Define the Azure Queue Storage ETL task:
// ========================================
var etlTask = new QueueEtlConfiguration()
{
  BrokerType = QueueBrokerType.AzureQueueStorage,
  
  Name = "myAzureQueueEtlTaskName",
  ConnectionStringName = "myAzureQueueConStr",
  
  Transforms = { transformation },
  
  // Set to false to allow task failover to another node if current one is down
  PinToMentorNode = false
}; 

// Deploy (send) the task to the server via the AddEtlOperation:
// =============================================================
store.Maintenance.Send(new AddEtlOperation<QueueConnectionString>(etlTask));

Delete processed documents:

  • You have the option to delete documents from your RavenDB database once they have been processed by the Queue ETL task.

  • Set the optional Queues property in your ETL configuration with the list of Azure queues for which processed documents should be deleted.

var etlTask = new QueueEtlConfiguration()
{
  BrokerType = QueueBrokerType.AzureQueueStorage,
  
  Name = "myAzureQueueEtlTaskName",
  ConnectionStringName = "myAzureQueueConStr",
  
  Transforms = { transformation },

  // Define whether to delete documents from RavenDB after they are sent to the target queue
  Queues = new List<EtlQueue>()
  {
      new()
      {
          // The name of the Azure queue  
          Name = "OrdersQueue",

          // When set to 'true',
          // documents that were processed by the transformation script will be deleted
          // from RavenDB after the message is loaded to the "OrdersQueue" in Azure.
          DeleteProcessedDocuments = true
      }
  }
}; 

store.Maintenance.Send(new AddEtlOperation<QueueConnectionString>(etlTask));

Syntax​

public class QueueEtlConfiguration
{
  // Set to QueueBrokerType.AzureQueueStorage to define an Azure Queue Storage ETL task
  public QueueBrokerType BrokerType { get; set; }
  // The ETL task name
  public string Name { get; set; }
  // The registered connection string name
  public string ConnectionStringName { get; set; }
  // List of transformation scripts
  public List<Transformation> Transforms { get; set; }
  // Optional configuration per queue
  public List<EtlQueue> Queues { get; set; }
  // Set to 'false' to allow task failover to another node if current one is down
  public bool PinToMentorNode { get; set; }
}

public class Transformation
{
  // The script name
  public string Name { get; set; }
  // The source RavenDB collections that serve as the input for the script
  public List<string> Collections { get; set; }
  // Set whether to apply the script on all collections
  public bool ApplyToAllDocuments { get; set; }
  // The script itself
  public string Script { get; set; }
}

public class EtlQueue
{
  // The Azure queue name
  public string Name { get; set; }
  // Delete processed documents when set to 'true'
  public bool DeleteProcessedDocuments { get; set; }
}

The transformation script​

The basic characteristics of an Azure Queue Storage ETL script are similar to those of other ETL types.
The script defines what data to extract from the source document, how to transform this data,
and which Azure Queue to load it to.

The loadTo method​

To specify which Azure queue to load the data into, use either of the following methods in your script.
The two methods are equivalent, offering alternative syntax:

  • loadTo<QueueName>(obj, {attributes})

    • Here the target is specified as part of the function name.
    • The target <QueueName> in this syntax is Not a variable and cannot be used as one,
      it is simply a string literal of the target's name.

  • loadTo('QueueName', obj, {attributes})

    • Here the target is passed as an argument to the method.
    • Separating the target name from the loadTo command makes it possible to include symbols like '-' and '.' in target names. This is not possible when the loadTo<QueueName> syntax is used because including special characters in the name of a JavaScript function makes it invalid.
    ParameterTypeDescription
    QueueNamestringThe name of the Azure Queue
    objobjectThe object to transfer
    attributesobjectAn object with optional & required CloudEvents attributes

For example, the following two calls, which load data to "OrdersQueue", are equivalent:

  • loadToOrdersQueue(obj, {attributes})
  • loadTo('OrdersQueue', obj, {attributes}) The following is a sample script that processes documents from the Orders collection:
// Create an orderData object
// ==========================
var orderData = {
  Id: id(this),
  OrderLinesCount: this.Lines.length,
  TotalCost: 0
};

// Update the orderData's TotalCost field
// ======================================
for (var i = 0; i < this.Lines.length; i++) {
  var line = this.Lines[i];
  var cost = (line.Quantity * line.PricePerUnit) * ( 1 - line.Discount);
  orderData.TotalCost += cost;
}

// Load the object to the "OrdersQueue" in Azure
// =============================================
loadToOrdersQueue(orderData, {
  Id: id(this),
  Type: 'com.example.promotions',
  Source: '/promotion-campaigns/summer-sale'
})

Note:
The queue name defined in the transform script must follow the set of rules outlined in:
Naming Queues and Metadata.

In this article