JSONL in Rust: serde_json, BufReader & Zero-Cost-Abstraktionen

use rayon::prelude::*;
use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader, BufWriter, Write};

#[derive(Debug, Deserialize, Serialize)]
struct Record {
    id: u64,
    text: String,
    #[serde(default)]
    word_count: Option<usize>,
}

fn parallel_process_jsonl(
    input_path: &str,
    output_path: &str,
) -> anyhow::Result<usize> {
    // Schritt 1: Alle Zeilen in den Speicher lesen
    let file = File::open(input_path)?;
    let reader = BufReader::new(file);
    let lines: Vec<String> = reader.lines().collect::<Result<_, _>>()?;

    // Schritt 2: Parallel parsen und transformieren
    let processed: Vec<Record> = lines
        .par_iter()
        .filter(|line| !line.trim().is_empty())
        .filter_map(|line| {
            serde_json::from_str::<Record>(line.trim()).ok()
        })
        .map(|mut record| {
            record.word_count = Some(record.text.split_whitespace().count());
            record
        })
        .collect();

    // Schritt 3: Ergebnisse schreiben
    let file = File::create(output_path)?;
    let mut writer = BufWriter::new(file);
    let count = processed.len();

    for record in &processed {
        serde_json::to_writer(&mut writer, record)?;
        writer.write_all(b"\n")?;
    }
    writer.flush()?;

    Ok(count)
}

fn main() -> anyhow::Result<()> {
    let count = parallel_process_jsonl("input.jsonl", "output.jsonl")?;
    println!("Processed {} records in parallel", count);
    Ok(())
}

use serde::Deserialize;
use std::fs::File;
use std::io::{BufRead, BufReader};
use thiserror::Error;

#[derive(Error, Debug)]
enum JsonlError {
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("JSON parse error at line {line}: {source}")]
    Parse {
        line: usize,
        source: serde_json::Error,
    },
    #[error("Validation error at line {line}: {message}")]
    Validation {
        line: usize,
        message: String,
    },
}

#[derive(Debug, Deserialize)]
struct Record {
    id: u64,
    name: String,
    value: f64,
}

fn validate_record(record: &Record, line: usize) -> Result<(), JsonlError> {
    if record.name.is_empty() {
        return Err(JsonlError::Validation {
            line,
            message: "name cannot be empty".into(),
        });
    }
    if record.value < 0.0 {
        return Err(JsonlError::Validation {
            line,
            message: format!("value must be non-negative, got {}", record.value),
        });
    }
    Ok(())
}

fn process_jsonl_with_errors(
    path: &str,
) -> Result<Vec<Record>, JsonlError> {
    let file = File::open(path)?; // Io-Fehler wird automatisch über #[from] konvertiert
    let reader = BufReader::new(file);
    let mut records = Vec::new();

    for (idx, line) in reader.lines().enumerate() {
        let line_num = idx + 1;
        let line = line?; // I/O-Fehler weitergeben
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }

        let record: Record = serde_json::from_str(trimmed)
            .map_err(|e| JsonlError::Parse { line: line_num, source: e })?;

        validate_record(&record, line_num)?;
        records.push(record);
    }

    Ok(records)
}

fn main() {
    match process_jsonl_with_errors("data.jsonl") {
        Ok(records) => println!("Successfully processed {} records", records.len()),
        Err(JsonlError::Io(e)) => eprintln!("File error: {}", e),
        Err(JsonlError::Parse { line, source }) => {
            eprintln!("JSON error at line {}: {}", line, source);
        }
        Err(JsonlError::Validation { line, message }) => {
            eprintln!("Validation error at line {}: {}", line, message);
        }
    }
}