Avro

use avro.qweruse avro_write, avro_read from avro      // Quillo 0.6: only these

apache-avro = "0.17"

// An export of page views as an Avro file for the data team, and the import that reads it.////   quillo run avro.qwe      then open http://localhost:8080/exportuse "avro.qwer"let SCHEMA = {    type: "record", name: "PageView", namespace: "shop.analytics",    fields: [        { name: "path", type: "string" },        { name: "visitor", type: "long" },        { name: "seconds", type: "double" },        { name: "country", type: ["null", "string"], default: null },        { name: "device", type: { type: "enum", name: "Device", symbols: ["DESKTOP", "MOBILE", "TABLET"] } },    ],}let VIEWS = [    { path: "/", visitor: 1001, seconds: 12.5, country: "NL", device: "DESKTOP" },    { path: "/products/b7", visitor: 1002, seconds: 48.0, country: null, device: "MOBILE" },    { path: "/checkout", visitor: 1001, seconds: 95.25, country: "NL", device: "DESKTOP" },]route GET /export {    let n = avro_write("data/views.avro", SCHEMA, VIEWS)?    { written: n, read_back: avro_read("data/views.avro", 0)?, schema_name: avro_schema("data/views.avro")?.name }}route GET /bad = { wrong_type: avro_write("data/bad.avro", SCHEMA, [{ path: "/", visitor: "not a number", seconds: 1.5, device: "DESKTOP" }]).error }

//! Apache Avro data files: rows written with a schema (as Kafka, Hadoop and BigQuery exports use) and read back with the schema they carry.//! version: 1.0.0//!//! [dependencies]//! apache-avro = "0.17"use apache_avro::{Codec, Reader, Schema, Writer};use serde_json::Value;/// At most a million rows are read at once.const MAX_ROWS: usize = 1_000_000;fn schema(s: &Value) -> Result<Schema, String> {    let text = match s {        Value::String(t) => t.clone(),        other => other.to_string(),    };    Schema::parse_str(&text).map_err(|e| format!("not an Avro schema: {}", e))}fn open(path: &str) -> Result<Reader<'static, std::io::BufReader<std::fs::File>>, String> {    let f = std::fs::File::open(path).map_err(|e| format!("{}: {}", path, e))?;    Reader::new(std::io::BufReader::new(f)).map_err(|e| format!("{}: not an Avro data file: {}", path, e))}/// Writes `rows` (maps) to a new Avro data file at `path` (folders made) with `avro_schema` (a map, or its JSON text),/// deflate-compressed; gives the number of rows. Each row must fit the schema: ints for `int` and `long`, numbers for/// `float` and `double`, a missing or null field only where the schema's union allows `null`. The error names the row.pub fn avro_write(path: &str, avro_schema: serde_json::Value, rows: Vec<serde_json::Value>) -> Result<i64, String> {    let s = schema(&avro_schema)?;    if let Some(dir) = std::path::Path::new(path).parent() {        if !dir.as_os_str().is_empty() {            std::fs::create_dir_all(dir).map_err(|e| format!("{}: {}", dir.display(), e))?;        }    }    let file = std::fs::File::create(path).map_err(|e| format!("{}: {}", path, e))?;    let mut w = Writer::with_codec(&s, std::io::BufWriter::new(file), Codec::Deflate);    for (i, row) in rows.iter().enumerate() {        let v = apache_avro::to_value(row).map_err(|e| format!("row {}: {}", i + 1, e))?;        let v = v.resolve(&s).map_err(|e| format!("row {} does not fit the schema: {}", i + 1, e))?;        w.append(v).map_err(|e| format!("row {}: {}", i + 1, e))?;    }    w.flush().map_err(|e| e.to_string())?;    w.into_inner().map_err(|e| e.to_string())?;    Ok(rows.len() as i64)}/// The rows of an Avro data file, as maps (at most `limit`, up to a million; 0 for the most). Enums are their names,/// unions their value, bytes lists of numbers, dates and timestamps numbers as Avro keeps them.pub fn avro_read(path: &str, limit: i64) -> Result<Vec<Value>, String> {    let max = if limit <= 0 { MAX_ROWS } else { (limit as usize).min(MAX_ROWS) };    let mut out = vec![];    for (i, v) in open(path)?.enumerate().take(max) {        let v = v.map_err(|e| format!("row {}: {}", i + 1, e))?;        out.push(Value::try_from(v).map_err(|e| format!("row {}: {}", i + 1, e))?);    }    Ok(out)}/// The schema an Avro data file was written with, as a map.pub fn avro_schema(path: &str) -> Result<Value, String> {    let r = open(path)?;    serde_json::to_value(r.writer_schema()).map_err(|e| e.to_string())}