Data Lake на Rust

Для чего нужна данная статья? :

Найти альтернативу реализации Data Lake на Rust в различных сценариях, связанных с обработкой и хранением больших объемов данных с использованием ML.

1. Обработка больших данных (Big Data)

• Потоковая обработка данных: для создания высокопроизводительных приложений для обработки потоков данных в реальном времени. Например, системы обработки данных в реальном времени (stream processing) могут обрабатывать большие объемы данных и хранить их в Data Lake.
• Параллельная обработка: для создания параллельных приложений, которые могут одновременно обрабатывать большие массивы данных, увеличивая производительность обработки в Data Lake.

2. ETL (Extract, Transform, Load)

• для написания высокопроизводительных инструментов для извлечения данных из различных источников, таких как базы данных, файлы, API и другие источники данных.
• для трансформации данных, например, для преобразования форматов данных, очистки данных или агрегирования информации перед загрузкой в Data Lake.
• для создания систем, которые загружают обработанные данные в Data Lake с минимальными задержками и высокой степенью надежности.

3. Аналитика данных

• Создание инструментов для анализа данных: для разработки высокопроизводительных аналитических инструментов, которые работают непосредственно с данными, хранящимися в Data Lake. Эти инструменты могут включать анализ данных, машинное обучение, генерацию отчетов и другие аналитические задачи.
• Интеграция с существующими системами: взаимодействие с другими аналитическими платформами или Data Lake, предоставляя интерфейсы для работы с данными или передавая данные на анализ.

4. Хранение и управление данными

• Создание эффективных форматов хранения: для создания кастомных форматов хранения данных в Data Lake, которые оптимизированы для конкретных задач, таких как быстрая выборка, компрессия или шифрование данных.

Зачем Вам это уметь? :

Создать быструю систему для хранения и обработки больших объемов данных.

Создадим основную структуру проекта на Rust:

cargo new data_lake

cd data_lake

В Cargo.toml добавьте зависимости для работы с различными компонентами:

[dependencies]

tokio = { version = "1", features = ["full"] }

serde = { version = "1.0", features = ["derive"] }

serde_json = "1.0"

rusoto_s3 = "0.47"

futures = "0.3"

Создадим модуль для загрузки данных в Data Lake, используя Amazon S3 в качестве хранилища:

use rusoto_core::{Region, RusotoError};

use rusoto_s3::{S3Client, S3, PutObjectRequest};

use tokio::fs::File;

use tokio::io::AsyncReadExt;

use std::path::Path;

use futures::stream::StreamExt;

pub async fn upload_to_s3(bucket: &str, key: &str, file_path: &str) ->

Result<(), RusotoError<rusoto_s3::PutObjectError>> {

let client = S3Client::new(Region::UsEast1);

let mut file = File::open(file_path).await.unwrap();

let mut buffer = Vec::new();

file.read_to_end(&mut buffer).await.unwrap();

let put_request = PutObjectRequest {

bucket: bucket.to_string(),

key: key.to_string(),

body: Some(buffer.into()),

..Default::default()

};

client.put_object(put_request).await?;

Ok(())

}
#[tokio::main]

async fn main() {

let bucket = "your-bucket-name";

let key = "data/file1.csv";

let file_path = "local/path/to/file1.csv";

match upload_to_s3(bucket, key, file_path).await {

Ok(_) => println!("File uploaded successfully!"),

Err(e) => println!("Error uploading file: {:?}", e),

}

}

Теперь создадим базовый ETL процесс, который загружает данные из CSV, трансформирует их и сохраняет обратно в Data Lake:

use tokio::fs::File;

use tokio::io::{AsyncBufReadExt, BufReader};

use serde::Deserialize;

#[derive(Debug, Deserialize)]

struct Record {

field1: String,
field2: i32,

field3: f64,

}

async fn process_csv(file_path: &str) -> Vec<Record> {

let file = File::open(file_path).await.unwrap();

let reader = BufReader::new(file);

let mut lines = reader.lines();

let mut records = Vec::new();

while let Some(line) = lines.next_line().await.unwrap() {

let record: Record = serde_json::from_str(&line).unwrap();

// Трансформация данных

let transformed_record = Record {

field1: record.field1.to_uppercase(),

field2: record.field2 * 2,

field3: record.field3 * 1.5,

};

records.push(transformed_record);

}

records

}
#[tokio::main]

async fn main() {

let file_path = "local/path/to/file.csv";

let transformed_data = process_csv(file_path).await;

// Далее эти данные можно загрузить обратно в Data Lake или обработать другим способом

for record in transformed_data {

println!("{:?}", record);

}

}

Data Lake с обработкой данных, машинным обучением для обнаружения аномалий, потоковой передачей через Kafka и REST API на Axum.

use std::sync::Arc;
use polars::prelude::*;

use rayon::prelude::*;

use arrow::record_batch::RecordBatch;

use tch::{nn, nn::Module, nn::OptimizerConfig, Device, Tensor};

use rdkafka::{producer::{BaseProducer, BaseRecord}, consumer::{StreamConsumer, Consumer}, ClientConfig};

use axum::{routing::get, Router};

use tokio::net::TcpListener;

use aws_sdk_s3::{Client as S3Client, Config as S3Config, Region};

use google_cloud_storage::client::{Client as GCPClient, ClientConfig as GCPConfig};

use azure_storage_blobs::prelude::*;

use moka::sync::Cache;

use prometheus::{Encoder, TextEncoder, Registry, opts, IntCounter};

use tracing::{info, error};

use tracing_subscriber;

// Глобальный кеш для хранения данных

lazy_static::lazy_static! {

static ref CACHE: Cache<String, DataFrame> = Cache::new(100);

static ref REQUEST_COUNTER: IntCounter = IntCounter::new("requests_total", "Total number of requests").unwrap();

static ref REGISTRY: Registry = Registry::new();

}

// Функция загрузки данных в Data Lake

fn load_data_lake(file_path: &str) -> Result<DataFrame> {

if let Some(cached_df) = CACHE.get(file_path) {

return Ok(cached_df);

}

let df = CsvReader::from_path(file_path)?.finish()?;

CACHE.insert(file_path.to_string(), df.clone());

Ok(df)

}

// Функция обработки данных с параллельной обработкой

fn process_data(df: &DataFrame) -> DataFrame {

df.clone()

.lazy()

.with_column(col("value").map(|s| s * 2, GetOutput::default()))

.collect()

.unwrap()

}

// Функция загрузки данных в облачное хранилище (AWS S3)

async fn upload_to_s3(bucket: &str, key: &str, data: &[u8]) {

let config = aws_config::load_from_env().await;

let client = S3Client::new(&config);

client.put_object().bucket(bucket).key(key).body(data.into()).send().await.unwrap();

info!("Uploaded data to AWS S3: {}/{}", bucket, key);

}

// Функция загрузки данных в облачное хранилище (GCP Storage)

async fn upload_to_gcp(bucket: &str, key: &str, data: &[u8]) {

let config = GCPConfig::default().with_auth().await.unwrap();

let client = GCPClient::new(config).unwrap();

client.upload(bucket, key, data).await.unwrap();

info!("Uploaded data to GCP Storage: {}/{}", bucket, key);

}

// Функция загрузки данных в облачное хранилище (Azure Blob Storage)

async fn upload_to_azure(container: &str, key: &str, data: &[u8]) {

let storage_account = std::env::var("AZURE_STORAGE_ACCOUNT").unwrap();

let storage_key = std::env::var("AZURE_STORAGE_KEY").unwrap();

let blob_client = StorageClient::new_access_key(&storage_account, &storage_key).blob_client(container, key);

blob_client.put_block_blob(data).await.unwrap();

info!("Uploaded data to Azure Blob Storage: {}/{}", container, key);

}

// Модель автоэнкодера для обнаружения аномалий

#[derive(Debug)]

struct Autoencoder {

encoder: nn::Sequential,

decoder: nn::Sequential,

}

impl Autoencoder {

fn new(vs: &nn::Path) -> Self {

let encoder = nn::seq().add(nn::linear(vs, 10, 5, Default::default()));

let decoder = nn::seq().add(nn::linear(vs, 5, 10, Default::default()));

Autoencoder { encoder, decoder }

}

fn forward(&self, xs: &Tensor) -> Tensor {

self.decoder.forward(&self.encoder.forward(xs))

}

}

// Kafka producer

fn send_to_kafka(topic: &str, key: &str, payload: &str) {

let producer: BaseProducer = ClientConfig::new()

.set("bootstrap.servers", "localhost:9092")

.create()

.expect("Producer creation error");

producer.send(BaseRecord::to(topic).key(key).payload(payload)).unwrap();

info!("Sent message to Kafka: {} - {}", topic, key);

}

// Kafka consumer

async fn consume_kafka(topic: &str) {

let consumer: StreamConsumer = ClientConfig::new()

.set("group.id", "data_lake")

.set("bootstrap.servers", "localhost:9092")

.create()

.expect("Consumer creation error");

consumer.subscribe(&[topic]).unwrap();

info!("Subscribed to Kafka topic: {}", topic);

}

// Функция для получения метрик Prometheus

async fn metrics() -> String {

let encoder = TextEncoder::new();

let mut buffer = Vec::new();

encoder.encode(&REGISTRY.gather(), &mut buffer).unwrap();

String::from_utf8(buffer).unwrap()

}

// API сервер

async fn api_server() {

let app = Router::new()

.route("/", get(|| async {

REQUEST_COUNTER.inc();

"Data Lake API"

}))

.route("/metrics", get(metrics));

let listener = TcpListener::bind("0.0.0.0:8080").await.unwrap();

info!("API server running on 0.0.0.0:8080");

axum::serve(listener, app).await.unwrap();

}

#[tokio::main]

async fn main() {

tracing_subscriber::fmt::init();

let file_path = "data.csv";

let df = load_data_lake(file_path).expect("Failed to load data");

let processed_df = process_data(&df);

let vs = nn::VarStore::new(Device::Cpu);

let model = Autoencoder::new(&vs.root());

let _result = model.forward(&Tensor::randn(&[10], (tch::Kind::Float, Device::Cpu)));

send_to_kafka("data_topic", "key1", "processed data");

consume_kafka("data_topic").await;

upload_to_s3("my-data-lake", "processed_data.parquet", b"binary_data").await;

upload_to_gcp("my-data-lake", "processed_data.parquet", b"binary_data").await;

upload_to_azure("my-container", "processed_data.parquet", b"binary_data").await;

api_server().await;

}