Queue ETL: Azure Queue Storage

Prior to setting up the ETL task, define a connection string that the task will use to access your Azure account.
The connection string includes the authorization credentials required to connect.

Authentication methods:

There are three authentication methods available:

• Connection string
  • Provide a single string that includes all the options required to connect to your Azure account.
    Learn more about Azure Storage connection strings here.
  • Note: the following connection string parameters are mandatory:
    • AccountName
    • AccountKey
    • DefaultEndpointsProtocol
    • QueueEndpoint (when using http protocol)
• Entra ID
  • Use the Entra ID authorization method to achieve enhanced security by leveraging Microsoft Entra’s robust identity solutions.
  • This approach minimizes the risks associated with exposed credentials commonly found in connection strings and enables more granular control through Role-Based Access Controls.
• Passwordless
  • This authorization method requires the machine to be pre-authorized and can only be used in self-hosted mode.
  • Passwordless authorization works only when the account on the machine is assigned the Storage Account Queue Data Contributor role; the Contributor role alone is inadequate.

Example:

// Prepare the connection string:
// ==============================
var conStr = new QueueConnectionString
{
    // Provide a name for this connection string
    Name = "myAzureQueueConStr", 
    
    // Set the broker type
    BrokerType = QueueBrokerType.AzureQueueStorage,
 
    // In this example we provide a simple string for the connection string
    AzureQueueStorageConnectionSettings = new AzureQueueStorageConnectionSettings()
    {
        ConnectionString = @"DefaultEndpointsProtocol=https;
                             AccountName=myAccountName;
                             AccountKey=myAccountKey;
                             EndpointSuffix=core.windows.net"
    }
};

// Deploy (send) the connection string to the server via the PutConnectionStringOperation:
// =======================================================================================
var res = store.Maintenance.Send(
    new PutConnectionStringOperation<QueueConnectionString>(conStr));

Syntax:

public 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; }
}

public enum QueueBrokerType
{
    None,
    Kafka,
    RabbitMq,
    AzureQueueStorage
}

public class AzureQueueStorageConnectionSettings
{
    public EntraId EntraId { get; set; }
    public string ConnectionString { get; set; }
    public Passwordless Passwordless { get; set; }
}

public class EntraId
{
    public string StorageAccountName { get; set; }
    public string TenantId { get; set; }
    public string ClientId { get; set; }
    public string ClientSecret { get; set; }
}

public class Passwordless
{
    public string StorageAccountName { get; set; }
}

// 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));

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 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.

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'
})