Rust原生HDFS客户端

hdfs-native是一个用Rust原生编写的HDFS客户端。它支持HDFS客户端几乎所有主要功能，以及下面列出的几个关键客户端配置选项。

支持的HDFS功能

以下是当前支持和未来可能支持但尚未支持的功能列表。

HDFS操作

• [x] 列表
• [x] 读取
• [x] 写入
• [x] 重命名
• [x] 删除
• [x] 基本权限和所有权
• [x] ACL
• [x] 内容摘要
• [x] 设置复制因子
• [x] 设置时间戳

HDFS特性

• [x] 名称服务
• [x] 观察者读取
• [x] ViewFS
• [x] 基于路由器的联邦
• [x] 纠删码读写
  • 仅支持RS模式，不支持RS-Legacy或XOR

安全特性

• [x] Kerberos认证（GSSAPI SASL支持）（需要libgssapi_krb5，见下文）
• [x] 令牌认证（DIGEST-MD5 SASL支持）
• [x] NameNode SASL连接
• [x] DataNode SASL连接
• [x] DataNode数据传输加密
• [ ] 静态加密（KMS支持）

Kerberos支持

Kerberos（SASL GSSAPI）机制通过运行时动态链接到libgssapi_krb5来支持。这必须单独安装，但可能已经安装在您的系统上。如果没有，您可以通过以下方式安装：

Debian系统

apt-get install libgssapi-krb5-2

RHEL系统

yum install krb5-libs

MacOS

brew install krb5

Windows

从下载并安装Microsoft Kerberos包

复制<INSTALL FOLDER>\MIT\Kerberos\bin\gssapi64.dll文件到%PATH%中的文件夹，并将名称更改为gssapi_krb5.dll

支持的HDFS设置

客户端将尝试在$HADOOP_CONF_DIR目录中读取Hadoop配置core-site.xml和hdfs-site.xml，如果该目录不存在，则在$HADOOP_HOME/etc/hadoop中读取。当前使用的支持的配置有：

• fs.defaultFS - Client::default()支持
• dfs.ha.namenodes - 名称服务支持
• dfs.namenode.rpc-address.* - 名称服务支持
• dfs.client.failover.resolve-needed.* - 基于DNS的NameNode发现
• dfs.client.failover.resolver.useFQDN.* - 基于DNS的NameNode发现
• dfs.client.failover.random.order.* - 随机化尝试NameNode的顺序
• dfs.client.failover.proxy.provider.* - 支持以下代理提供程序的行为。任何其他值将默认回退到ConfiguredFailoverProxyProvider行为：
  • org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
  • org.apache.hadoop.hdfs.server.namenode.ha.ObserverReadProxyProvider
  • org.apache.hadoop.hdfs.server.namenode.ha.RouterObserverReadConfiguredFailoverProxyProvider
• dfs.client.block.write.replace-datanode-on-failure.enable
• dfs.client.block.write.replace-datanode-on-failure.policy
• dfs.client.block.write.replace-datanode-on-failure.best-effort
• fs.viewfs.mounttable.*.link.* - ViewFS链接
• fs.viewfs.mounttable.*.linkFallback - ViewFS链接回退

所有其他设置通常假定为当前默认值。例如，安全性被假定为启用，并且始终进行SASL协商，但在不安全的集群上，这将仅进行SIMPLE认证。需要其他自定义Hadoop客户端配置的任何设置可能无法正常工作。

构建

cargo build

对象存储实现

HDFS的对象存储实现在hdfs-native-object-store crate中提供。

运行测试

测试主要是集成测试，利用rust/mindifs/中的一个小型Java应用程序运行自定义MiniDFSCluster。要运行测试，您需要安装Java、Maven、Hadoop二进制文件和Kerberos工具，并在您的路径上可用。任何Java版本8到17之间都应该可以工作。

cargo test -p hdfs-native --features intergation-test

Python测试

请参阅Python README

运行基准测试

一些基准测试通过fs-hdfs3 crate通过libhdfs与基于JVM的客户端进行性能比较。因此，运行基准测试需要一些额外的设置：

export HADOOP_CONF_DIR=$(pwd)/rust/target/test
export CLASSPATH=$(hadoop classpath)

然后您可以使用以下命令运行基准测试：

cargo bench -p hdfs-native --features benchmark

benchmark功能需要暴露minidfs和内部纠删码功能以进行基准测试。

运行示例

示例使用minidfs模块创建一个简单的HDFS集群来运行示例。这需要包含integration-test功能以启用minidfs模块。或者，如果您想针对现有的HDFS集群运行示例，可以排除integration-test功能，并确保您的HADOOP_CONF_DIR指向包含与您的集群通信的HDFS配置的目录。

cargo run --example simple --features integration-test

完整示例代码

use hdfs_native::Client;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 创建HDFS客户端
    let client = Client::default()?;
    
    // 列出根目录下的文件
    let entries = client.list_status("/").await?;
    for entry in entries {
        println!("Name: {}, Type: {:?}", entry.path, entry.file_type);
    }
    
    // 创建并写入文件
    let path = "/test_file.txt";
    let mut writer = client.create(path).await?;
    writer.write_all(b"Hello, HDFS from Rust!").await?;
    writer.close().await?;
    
    // 读取文件内容
    let mut reader = client.read(path).await?;
    let mut content = Vec::new();
    reader.read_to_end(&mut content).await?;
    println!("File content: {}", String::from_utf8_lossy(&content));
    
    // 删除文件
    client.delete(path, false).await?;
    
    Ok(())
}

use hdfs_native::Client;
use std::io::Cursor;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 使用特定配置创建客户端
    let client = Client::new("hdfs://namenode:8020")?;
    
    // 创建目录
    client.mkdirs("/test_directory", 0o755).await?;
    
    // 写入多个文件
    for i in 0..5 {
        let file_path = format!("/test_directory/file_{}.txt", i);
        let content = format!("This is file number {}", i);
        
        let mut writer = client.create(&file_path).await?;
        writer.write_all(content.as_bytes()).await?;
        writer.close().await?;
    }
    
    // 递归列出目录内容
    let entries = client.list_status_recursive("/test_directory").await?;
    for entry in entries {
        println!("Path: {}, Size: {} bytes", entry.path, entry.length);
    }
    
    // 重命名目录
    client.rename("/test_directory", "/renamed_directory").await?;
    
    // 删除目录
    client.delete("/renamed_directory", true).await?;
    
    Ok(())
}

use hdfs_native::Client;
use tokio::io::AsyncWriteExt;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 创建支持Kerberos认证的客户端
    let client = Client::default()?;
    
    // 检查文件是否存在
    let test_file = "/large_file.dat";
    if client.exists(test_file).await? {
        println!("File already exists, deleting...");
        client.delete(test_file, false).await?;
    }
    
    // 创建大文件并分块写入
    let mut writer = client.create(test_file).await?;
    let chunk_size = 1024 * 1024; // 1MB chunks
    let total_size = 10 * 1024 * 1024; // 10MB total
    
    for i in 0..(total_size / chunk