The Azure Blob Storage sink connector pulls data from Pulsar topics and persists data to Azure Blob Storage containers.

Quick start

Prerequisites

The prerequisites for connecting an Azure Blob Storage sink connector to external systems include:

1. Create Blob Storage container in Azure Cloud.
2. Get Storage account Connection string.

1. Create a connector

The following command shows how to use pulsarctl to create a builtin connector. If you want to create a non-builtin connector, you need to replace --sink-type cloud-storage-azure-blob with --archive /path/to/pulsar-io-cloud-storage.nar. You can find the button to download the nar package at the beginning of the document.

pulsarctl sinks create \
  --sink-type cloud-storage-azure-blob \
  --name azure-blob-sink \
  --tenant public \
  --namespace default \
  --inputs "Your topic name" \
  --parallelism 1 \
  --sink-config \
  '{
    "azureStorageAccountConnectionString": "Your azure blob storage account connection string",
    "provider": "azure-blob-storage",
    "bucket": "Your container name",
    "formatType": "json",
    "partitionerType": "PARTITION"
  }'

The --sink-config is the minimum necessary configuration for starting this connector, and it is a JSON string. You need to substitute the relevant parameters with your own. If you want to configure more parameters, see Configuration Properties for reference.

2. Send messages to the topic

    public static void main(String[] args) throws Exception {
        PulsarClient client = PulsarClient.builder()
                             .serviceUrl("{{Your Pulsar URL}}")
                             .build();


        Producer<String> producer = client.newProducer(Schema.STRING)
                .topic("{{Your topic name}}")
                .create();

        for (int i = 0; i < 10; i++) {
            // JSON string containing a single character
            String message = "{\"test-message\": \"test-value\"}";
            producer.send(message);
        }

        producer.close();
        client.close();
    }

3. Display data on Azure Blob Storage console

You can see the object at public/default/{{Your topic name}}-partition-0/xxxx.json on the Azure Blob Storage console. Download and open it, the content is:

{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}
{"test-message":"test-value"}

Configuration Properties

Before using the Azure Blob Storage sink connector, you need to configure it. This table outlines the properties and the descriptions.

There are three methods to authenticate with Azure Blob Storage:

1. azureStorageAccountConnectionString: This method involves using the Azure Blob Storage connection string for authentication. It's the simplest method as it only requires the connection string.
2. azureStorageAccountSASToken: This method uses a Shared Access Signature (SAS) token for the Azure Blob Storage account. When using this method, you must also set the endpoint.
3. azureStorageAccountName and azureStorageAccountKey: This method uses the Azure Blob Storage account name and account key for authentication. Similar to the SAS token method, you must also set the endpoint when using this method.

Advanced features

Data format types

Azure Blob Storage Sink Connector provides multiple output format options, including JSON, Avro, Bytes, or Parquet. The default format is JSON. With current implementation, there are some limitations for different formats:

This table lists the Pulsar Schema types supported by the writers.

*: The JSON writer will try to convert the data with a String or Bytes schema to JSON-format data if convertable.

**: The Protobuf schema is based on the Avro schema. It uses Avro as an intermediate format, so it may not provide the best effort conversion.

***: The ProtobufNative record holds the Protobuf descriptor and the message. When writing to Avro format, the connector uses avro-protobuf to do the conversion.

This table lists the support of withMetadata configurations for different writer formats:

*: When using Parquet with PROTOBUF_NATIVE format, the connector will write the messages with DynamicMessage format. When withMetadata is set to true, the connector will add __message_metadata__ to the messages with PulsarIOCSCProtobufMessageMetadata format.

For example, if a message User has the following schema:

syntax = "proto3";
message User {
 string name = 1;
 int32 age = 2;
}

When withMetadata is set to true, the connector will write the message DynamicMessage with the following schema:

syntax = "proto3";
message PulsarIOCSCProtobufMessageMetadata {
 map<string, string> properties = 1;
 string schema_version = 2;
 string message_id = 3;
}
message User {
 string name = 1;
 int32 age = 2;
 PulsarIOCSCProtobufMessageMetadata __message_metadata__ = 3;
}

Dead-letter topics

To use a dead-letter topic, you need to set skipFailedMessages to false, and set --max-redeliver-count and --dead-letter-topic when submit the connector with the pulsar-admin CLI tool. For more info about dead-letter topics, see the Pulsar documentation. If a message fails to be sent to the Azure Blob Storage and there is a dead-letter topic, the connector will send the message to the dead-letter topic.

Sink flushing only after batchTimeMs elapses

There is a scenario where the sink is only flushing whenever the batchTimeMs has elapsed, even though there are many messages waiting to be processed. The reason for this is that the sink will only acknowledge messages after they are flushed to the Azure Blob Storage but the broker stops sending messages when it reaches a certain limit of unacknowledged messages. If this limit is lower or close to batchSize, the sink never receives enough messages to trigger a flush based on the amount of messages. In this case please ensure the maxUnackedMessagesPerConsumer set in the broker configuration is sufficiently larger than the batchSize setting of the sink.

Partitioner Type

There are two types of partitioner:

• PARTITION: This is the default partitioning method based on Pulsar partitions. In other words, data is partitioned according to the pre-existing partitions in Pulsar topics. For instance, a message for the topic public/default/my-topic-partition-0 would be directed to the file public/default/my-topic-partition-0/xxx.json, where xxx signifies the earliest messageId(Format: ledgerId.entryId.batchIndex)/offset(Enable config: partitionerUseIndexAsOffset) in this file.

• TIME: Data is partitioned according to the time it was flushed. Using the previous message as an example, if it was received on 2023-12-20, it would be directed to public/default/my-topic-partition-0/2023-12-20/xxx.json, where xxx also denotes the earliest messageId(Format: ledgerId.entryId.batchIndex)/offset(Enable config: partitionerUseIndexAsOffset) in this file.