Rust AWS数据传输服务库aws-sdk-firehose的使用指南

概述

aws-sdk-firehose是AWS官方提供的Rust SDK，用于与Amazon Kinesis Data Firehose服务交互。Firehose是一项完全托管的服务，用于将实时流数据可靠地加载到数据存储和分析服务中。

主要功能

• 将流数据可靠地传输到Amazon S3、Redshift、Elasticsearch等服务
• 自动处理数据转换、压缩和加密
• 近实时数据传输(60秒内)
• 自动扩展以匹配数据吞吐量

安装与配置

首先在Cargo.toml中添加依赖：

[dependencies]
aws-config = "0.55"
aws-sdk-firehose = "0.28"
tokio = { version = "1", features = ["full"] }

基本使用方法

1. 创建Firehose客户端

use aws_sdk_firehose::Client;

async fn create_client() -> Client {
    let config = aws_config::load_from_env().await;
    Client::new(&config)
}

2. 发送记录到Firehose

use aws_sdk_firehose::model::Record;

async fn put_record(client: &Client, delivery_stream_name: &str, data: &str) {
    let record = Record::builder()
        .data(aws_sdk_firehose::Blob::new(data))
        .build();
    
    let response = client
        .put_record()
        .delivery_stream_name(delivery_stream_name)
        .record(record)
        .send()
        .await
        .unwrap();
    
    println!("Record ID: {:?}", response.record_id);
}

3. 批量发送记录

async fn put_record_batch(client: &Client, delivery_stream_name: &str, records: Vec<&str>) {
    let records: Vec<Record> = records
        .into_iter()
        .map(|data| {
            Record::builder()
                .data(aws_sdk_firehose::Blob::new(data))
                .build()
        })
        .collect();
    
    let response = client
        .put_record_batch()
        .delivery_stream_name(delivery_stream_name)
        .set_records(Some(records))
        .send()
        .await
        .unwrap();
    
    println!(
        "Successfully sent {} records, failed {}",
        response.successful_count(),
        response.failed_put_count()
    );
}

高级用法示例

1. 创建新的传输流

use aws_sdk_firehose::model::{S3DestinationConfiguration, ExtendedS3DestinationConfiguration};

async fn create_delivery_stream(client: &Client, stream_name: &str, bucket_arn: &str) {
    let s3_config = S3DestinationConfiguration::builder()
        .bucket_arn(bucket_arn)
        .role_arn("arn:aws:iam::123456789012:role/firehose_delivery_role")
        .build();
    
    let response = client
        .create_delivery_stream()
        .delivery_stream_name(stream_name)
        .s3_destination_configuration(s3_config)
        .send()
        .await
        .unwrap();
    
    println!("Delivery stream ARN: {:?}", response.delivery_stream_arn);
}

2. 更新现有传输流配置

async fn update_destination(client: &Client, stream_name: &str, bucket_arn: &str) {
    let extended_s3_config = ExtendedS3DestinationConfiguration::builder()
        .bucket_arn(bucket_arn)
        .role_arn("arn:aws:iam::123456789012:role/firehose_delivery_role")
        .buffering_hints(
            BufferingHints::builder()
                .interval_in_seconds(60)
                .size极管理、监控和扩展。

## 完整示例代码

```rust
use aws_sdk_firehose::{Client, Error};
use aws_sdk_firehose::model::Record;
use chrono::Utc;

#[tokio::main]
async fn main() -> Result<(), Error> {
    // 创建Firehose客户端
    let client = create_client().await;
    let stream_name = "my-data-stream";
    
    // 生成测试数据 - 10条JSON格式记录
    let records = (0..10)
        .map(|i| format!("{{\"id\": {}, \"timestamp\": \"{}\"}}", i, Utc::now()))
        .collect::<Vec<_>>();
    
    // 批量发送记录
    put_record_batch(&client, stream_name, records).await;
    
    Ok(())
}

// 创建Firehose客户端
async fn create_client() -> Client {
    // 从环境变量加载AWS配置
    let config = aws_config::load_from_env().await;
    Client::new(&config)
}

// 批量发送记录到Firehose
async fn put_record_batch(client: &Client, delivery_stream_name: &str, records: Vec<String>) {
    // 将字符串数据转换为Record对象
    let records: Vec<Record> = records
        .into_iter()
        .map(|data| {
            Record::builder()
                .data(aws_sdk_firehose::Blob::new(data))
                .build()
        })
        .collect();
    
    // 发送批量记录并处理响应
    match client
        .put_record_batch()
        .delivery_stream_name(delivery_stream_name)
        .set_records(Some(records))
        .send()
        .await
    {
        Ok(response) => {
            // 打印发送结果统计
            println!(
                "成功发送 {} 条记录，失败 {} 条",
                response.successful_count(),
                response.failed_put_count()
            );
            
            // 打印失败记录的详细信息
            if let Some(failed_records) = response.request_responses {
                for record in failed_records {
                    println!("失败记录: {:?}", record.error_message);
                }
            }
        }
        Err(e) => eprintln!("发送记录时出错: {}", e),
    }
}

这个完整示例展示了如何：

1. 创建Firehose客户端
2. 生成测试数据
3. 批量发送数据到Firehose
4. 处理发送结果和错误

通过这个示例，开发者可以快速上手使用aws-sdk-firehose构建实时数据传输应用。