- Configure Private Cloud
- Private Preview
- Lakehouse Tiered Storage
User Guide
Note
This feature is currently in private preview. If you want to try it out or have any questions, submit a ticket to the support team.
Overview
This tutorial offers a detailed guide on configuring Lakehouse tiered storage within a private cloud environment.
Supported Features
The Lakehouse tiered storage supports the following features:
- Offload storage options: HDFS, and Delta
- Data production and consumption with AVRO schema and Pulsar primitive schema
- Prerequisites: JDK 17+ and Pulsar 3.0.0+
Configuration Steps
To set up Lakehouse tiered storage, create an offload.conf file in the conf directory. The configuration parameters include:
Offload Framework Configurations
| Configuration | Description | Required | Default Value |
|---|---|---|---|
| metadataServiceUri | The metadata service uri for bookkeeper client, for example, zk://localhost:2181/ledgers | Y | N/A |
| pulsarWebServiceUrl | The pulsar web service URL, for example, http://localhost:8080 | Y | N/A |
| pulsarServiceUrl | The pulsar protocol service URL, for example, pulsar://localhost:6650 | Y | N/A |
| offloadProvider | The offloader driver's name, for example, delta or iceberg | Y | N/A |
| maxContainerCountPerTopic | The maximum containers will be used by a single topic. | N | 2 |
| containerSizeInBytes | The maximum bytes for each container. By default, the container can fill with 24 MB messages to write. | N | 25165824 |
| enableSharedMessageContainer | Enable shared message container or not. Once the shared message container is enabled, all the offload processors will get the container from the shared message container pool | N | true |
| sharedMessageContainerPoolMemory | The max shared message container pool memory size. Default: 0.125 * maxDirectMemorySize | N | 0.125 * maxDirectMemory |
| offloadSubName | offload subscription name | N | __OFFLOAD |
| initialDelayTime | The initial delay time in milliseconds | N | 10000 |
| maxRetryTimes | The max retry times of the delay, when the backoff reaches the max retry number, the delay time won't increase anymore | N | 5 |
| maxContainerSwitchTime | The max time that message container waiting to switch. Time unit: second. Default: 60 seconds. | N | 60 |
| maxReadCacheSize | Amount of memory to use for caching data payload in offload framework. This memory is allocated from JVM direct memory and it's shared across all the topics running in the same broker. By default, uses 1/8th of available direct memory. | N | Math.max(64MB, (int) (0.125 * maxDirectMemory())) |
| readContainerBatchSize | The bytes size of entries one read container hold to read. Default is 5MB. | N | 5242880 |
| offloadWriterMaxRetryTimes | Offload writer the max retry times of the delay to retry for retryable write error. | N | 3 |
| embeddedStatServiceEnabled | Enable the embedded stat service | N | true |
| statServicePort | Stat service default port | N | 8090 |
| offloadWriterThreads | The number of thread pool threads for lakehouse writer | N | 2 |
| offloadReaderThreads | The number of thread pool threads for Lakehouse reader | N | 2 |
| offloadCoreThreads | The number of thread pool threads for the core processor | N | 2 |
| offloadHighPriorityTaskThreads | Maximum number of thread pool threads for the core processor high-priority task | N | 2 |
| cluster | The tiered storage offloading cluster name, and it will be shown on the Prometheus metrics label. If you run the offloader within the Pulsar broker, you can set it the same with the Pulsar cluster name. Default: pulsar | N | pulsar |
| schemaRegistryType | The schema registry type, we support 'pulsar','kop','fisher', default is 'pulsar'. | N | pulsar |
| messageFormat | The msg's format, we support 'avro','json','csv','kop', default is 'avro'. | N | avro |
| skipDirtyData | Whether to skip the dirty data, if skip, it will also ack the msg although encode failed. | N | false |
| appendPulsarMetaField | Whether to append pulsar meta filed(such as: **ledgerId, **entryId). If disabled, can't read the msg from tiered storage. | N | true |
| partitionKey | The custom partition key in the lakehouse table, support multi partition key split by ','. It only works when 'appendPulsarMetaField' is disabled. Default value is 'none' | N | none |
| dataManagedByPulsar | Whether the data lifecycle is managed by Pulsar or not. If setting it to false, all the delete request from Pulsar won't be processed to the tiered storage | N | true |
Offload Writer/Reader Configurations
| Configuration | Description | Required | Default Value |
|---|---|---|---|
| storagePath | The data stored path. For example, gs://offload-dev (GCS offload) or s3a://offload-dev (S3 offload) | N | data |
| hdfsResourceFile | Hadoop profile path. The configuration file is stored in the Hadoop profile path. It contains various settings for Hadoop performance tuning. | N | conf/filesystem_offload_core_site.xml |
| maxCommitInterval | The max commit interval before committing data to the Lakehouse table. Set it to -1 to disable the time-based flush. Default: -1 | N | -1 |
| maxCommitSize | The max commits size before committing data to the Lakehouse table. Set it to -1 to disable the message size based flush. Default: -1 | N | -1 |
| googleCloudProjectID | GCS offload project configuration. For example: example-project | Required if offloading data to GCS | N/A |
| googleCloudServiceAccountFile | GCS offload authentication. For example: /Users/user-name/Downloads/project-804d5e6a6f33.json | Required if offloading data to GCS | N/A |
| hadoop.xxx | All the Hadoop-related configurations can be set with hadoop. prefix | N | N/A |
| delta.compressType | The compression type for the delta table, Available options are: UNCOMPRESSED, SNAPPY, GZIP, LZO, BROTLI, LZ4, ZSTD. Default: SNAPPY | N | SNAPPY |
| delta.fileType | The file type for the delta table, Available options are: PARQUET. Default: PARQUET | N | PARQUET |
| delta.rowGroupSize | The row group size for the delta table. Default: 10MB | N | 10485760 |
| delta.maxLedgerNumOnOneIndex | The delta max ledger num on one index, default is 1024. | N | 1024 |
| delta.reader.count | The max opened delta reader count, default is 50. | N | 50 |
| delta.compactThresholdSize | The delta compact threshold size, default is 100MB. | N | 104857600 |
If you want to offload data to S3, you need to export AWS_ACCESS_KEY_ID and AWS_SECRET_KEY before starting up the broker service and setting the storagePath with s3a:// prefix.
Limitations
Framework limitations include:
- Schema evolution does not support deletion
- Supported schema types vary for different offloaders
- AVRO and Pulsar primitive schema (Delta offloader)
- All the schema types for HDFS offloader
- Inability to configure different drivers for distinct namespaces/topics
Delta-specific limitations:
- Avoid using Spark for small file compaction to prevent interference with delta streaming reads. Separate support for the compaction feature is available.
Notice
In Lakehouse tiered storage, the RawReader is utilized to retrieve messages from Pulsar topics and write them to the Lakehouse. An essential component of this process is the offload cursor, which marks the progress of the offloading operation. It's crucial to note that advancing the offload cursor prematurely can lead to data loss.
When considering the configuration of Time-to-Live (TTL) settings at the namespace or topic level, there are two key options to adhere to:
- Do Not Configure TTL at Namespace or Topic Level:
- This option ensures that no TTL constraints are imposed at the namespace or topic level, allowing data to persist without automatic expiration based on time.
- Configure TTL at Namespace or Topic Level:
- If TTL settings are implemented at the namespace or topic level, it is imperative to ensure that the TTL value is greater than the retention policy set for the topic or namespace.
By carefully considering these options and aligning TTL configurations with retention policies, organizations can effectively manage data lifecycle and retention within the Lakehouse tiered storage environment, safeguarding against unintended data loss scenarios.
By following these configurations and considerations, you can effectively set up and manage Lakehouse tiered storage in your private cloud environment.