Apache OpenDAL™ Parquet 集成

parquet_opendal 提供了 parquet 高效的 IO 工具。

有用链接

• 文档： 发布 | 开发

示例

将以下依赖项添加到您的 Cargo.toml 中，并使用正确的版本

[dependencies]
parquet_opendal = "0.0.1"
opendal = { version = "0.48.0", features = ["services-s3"] }

use std::sync::Arc;

use arrow::array::{ArrayRef, Int64Array, RecordBatch};

use futures::StreamExt;
use opendal::{services::S3Config, Operator};
use parquet::arrow::{AsyncArrowWriter, ParquetRecordBatchStreamBuilder};
use parquet_opendal::{AsyncReader, AsyncWriter};

#[tokio::main]
async fn main() {
    let mut cfg = S3Config::default();
    cfg.access_key_id = Some("my_access_key".to_string());
    cfg.secret_access_key = Some("my_secret_key".to_string());
    cfg.endpoint = Some("my_endpoint".to_string());
    cfg.region = Some("my_region".to_string());
    cfg.bucket = "my_bucket".to_string();

    // Create a new operator
    let operator = Operator::from_config(cfg).unwrap().finish();
    let path = "/path/to/file.parquet";

    // Create an async writer
    let writer = AsyncWriter::new(
        operator
            .writer_with(path)
            .chunk(32 * 1024 * 1024)
            .concurrent(8)
            .await
            .unwrap(),
    );

    let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
    let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
    let mut writer = AsyncArrowWriter::try_new(writer, to_write.schema(), None).unwrap();
    writer.write(&to_write).await.unwrap();
    writer.close().await.unwrap();

    /// `gap(512 * 1024)` - Sets the maximum gap size (in bytes) to merge small byte ranges
    ///   to 512 KB.
    /// `chunk(16 * 1024 * 1024)` - Sets the chunk size (in bytes) for reading data to 16 MB.
    /// `concurrent(16)` - Sets the number of concurrent fetch operations to 16.
    let reader = operator
        .reader_with(path)
        .gap(512 * 1024)
        .chunk(16 * 1024 * 1024)
        .concurrent(16)
        .await
        .unwrap();

    let content_len = operator.stat(path).await.unwrap().content_length();
    // `with_prefetch_footer_size(512 * 1024)` - Sets the prefetch footer size to 512 KB.
    let reader = AsyncReader::new(reader, content_len).with_prefetch_footer_size(512 * 1024);
    let mut stream = ParquetRecordBatchStreamBuilder::new(reader)
        .await
        .unwrap()
        .build()
        .unwrap();
    let read = stream.next().await.unwrap().unwrap();
    assert_eq!(to_write, read);
}

品牌

第一次和最突出的提及必须使用完整形式：Apache OpenDAL™ 的名称用于任何个人使用（网页、手册、幻灯片等）。根据上下文和写作风格，您应该足够频繁地使用名称的完整形式，以确保读者清楚地了解 OpenDAL 项目和 OpenDAL 软件产品与 ASF（Apache 软件基金会）作为母组织之间的关联。

有关更多详细信息，请参阅Apache 产品名称使用指南。

许可和商标

根据 Apache 许可协议版本 2.0 许可： https://apache.ac.cn/licenses/LICENSE-2.0

Apache OpenDAL、OpenDAL 和 Apache 是 Apache 软件基金会的注册商标或商标。