sink
Google Cloud Storage Sink Connector
Cloud Storage Connector integrates Apache Pulsar with cloud storage.

Available on
StreamNative Cloud console

Authored by
StreamNative
Support type
streamnative
License
Apache License 2.0

The Google Cloud Storage sink connector pulls data from Pulsar topics and persists data to Google Cloud Storage buckets.

Quick start

Prerequisites

The prerequisites for connecting an Google Cloud Storage sink connector to external systems include:

  1. Create Cloud Storage buckets in Google Cloud.
  2. Create the Google cloud ServiceAccount and create a public key certificate.
  3. Create the Google cloud Role, ensure the Google Cloud role have the following permissions:
- storage.buckets.get
- storage.buckets.list
- storage.objects.create
  1. Grant the ServiceAccount the above Role.

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

For StreamNative Cloud User

If you are a StreamNative Cloud user, you need set up your environment first.

pulsarctl sinks create \
  --sink-type cloud-storage-gcloud \
  --name gcloud-storage-sink \
  --tenant public \
  --namespace default \
  --inputs "Your topic name" \
  --parallelism 1 \
  --sink-config \
  '{
    "gcsServiceAccountKeyFileContent": "Public key certificate you created above", 
    "provider": "google-cloud-storage",
    "bucket": "Your bucket 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.

Note

You can also choose to use a variety of other tools to create a connector:

2. Send messages to the topic

Note

If your connector is created on StreamNative Cloud, you need to authenticate your clients. See Build applications using Pulsar clients for more information.

    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 Google Cloud Storage console

You can see the object at public/default/{{Your topic name}}-partition-0/xxxx.json on the Google Cloud 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 Google Cloud Storage sink connector, you need to configure it. This table outlines the properties and the descriptions.

NameTypeRequiredDefaultDescription
providerStringTruenullThe Cloud Storage type, google cloud storage only supports the google-cloud-storage provider.
bucketStringTruenullThe Cloud Storage bucket.
formatTypeStringTruenullThe data format type. Available options are JSON, Avro, Bytes, or Parquet. By default, it is set to JSON.
partitionerTypeStringTruenullThe partitioning type. It can be configured by topic partitions or by time. By default, the partition type is configured by topic partitions.
gcsServiceAccountKeyFileContentStringFalse""The contents of the JSON service key file. If empty, credentials are read from gcsServiceAccountKeyFilePath file.
gcsServiceAccountKeyFilePathStringFalse""Path to the GCS credentials file. If empty, the credentials file will be read from the GOOGLE_APPLICATION_CREDENTIALS environment variable.
timePartitionPatternStringFalse"yyyy-MM-dd"The format pattern of the time-based partitioning. For details, refer to the Java date and time format.
timePartitionDurationStringFalse"86400000"The time interval for time-based partitioning. Support formatted interval string, such as 30d, 24h, 30m, 10s, and also support number in milliseconds precision, such as 86400000 refers to 24h or 1d.
partitionerUseIndexAsOffsetBooleanFalsefalseWhether to use the Pulsar's message index as offset or the record sequence. It's recommended if the incoming messages may be batched. The brokers may or not expose the index metadata and, if it's not present on the record, the sequence will be used. See PIP-70 for more details.
batchSizeintFalse10The number of records submitted in batch.
batchTimeMslongFalse1000The interval for batch submission.
maxBatchByteslongFalse10000000The maximum number of bytes in a batch.
sliceTopicPartitionPathBooleanFalsefalseWhen it is set to true, split the partitioned topic name into separate folders in the bucket path.
withMetadataBooleanFalsefalseSave message attributes to metadata.
useHumanReadableMessageIdBooleanFalsefalseUse a human-readable format string for messageId in message metadata. The messageId is in a format like ledgerId:entryId:partitionIndex:batchIndex. Otherwise, the messageId is a Hex-encoded string.
withTopicPartitionNumberBooleanFalsetrueWhen it is set to true, include the topic partition number to the object path.
bytesFormatTypeSeparatorStringFalse"0x10"It is inserted between records for the formatType of bytes. By default, it is set to '0x10'. An input record that contains the line separator looks like multiple records in the output object.
pendingQueueSizeintFalse10The number of records buffered in queue. By default, it is equal to batchSize. You can set it manually.
useHumanReadableSchemaVersionBooleanFalsefalseUse a human-readable format string for the schema version in the message metadata. If it is set to true, the schema version is in plain string format. Otherwise, the schema version is in hex-encoded string format.
skipFailedMessagesBooleanFalsefalseConfigure whether to skip a message which it fails to be processed. If it is set to true, the connector will skip the failed messages by ack it. Otherwise, the connector will fail the message.
pathPrefixStringFalsefalseIf it is set, the output files are stored in a folder under the given bucket path. The pathPrefix must be in the format of xx/xxx/.
avroCodecStringFalsesnappyCompression codec used when formatType=avro. Available compression types are: null (no compression), deflate, bzip2, xz, zstandard, snappy.
parquetCodecStringFalsegzipCompression codec used when formatType=parquet. Available compression types are: null (no compression), snappy, gzip, lzo, brotli, lz4, zstd.
jsonAllowNaNBooleanFalsefalseRecognize 'NaN', 'INF', '-INF' as legal floating number values when formatType=json. Since JSON specification does not allow such values this is a non-standard feature and disabled by default.

Advanced features

Data format types

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

Pulsar SchemaWriter: AvroWriter: JSONWriter: ParquetWriter: Bytes
Primitive✗✔ *✗✔
Avro✔✔✔✔
Json✔✔✔✔
Protobuf **✔✔✔✔
ProtobufNative✔ ***✗✔✔

*: 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:

Writer FormatwithMetadata
Avro✔
JSON✔
Parquet✔ *
Bytes✗

*: 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 Cloud 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 cloud 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.