static USAGE: &str = r#" Convert newline-delimited JSON (JSONL/NDJSON) to CSV. The command tries to do its best but since it is not possible to straightforwardly convert JSON lines to CSV, the process might lose some complex fields from the input. Also, it will fail if the JSON documents are not consistent with one another, as the first JSON line will be used to infer the headers of the CSV output. For examples, see https://github.com/dathere/qsv/blob/master/tests/test_jsonl.rs. Usage: qsv jsonl [options] [] qsv jsonl --help jsonl options: --ignore-errors Skip malformed input lines. -j, --jobs The number of jobs to run in parallel. When not set, the number of jobs is set to the number of CPUs detected. -b, --batch The number of rows per batch to load into memory, before running in parallel. Set to 0 to load all rows in one batch. [default: 50000] Common options: -h, --help Display this message -o, --output Write output to instead of stdout. -d, --delimiter The delimiter to use when writing CSV data. Must be a single character. [default: ,] "#; use std::{ fs, io::{self, BufRead, BufReader}, }; use rayon::{ iter::{IndexedParallelIterator, ParallelIterator}, prelude::IntoParallelRefIterator, }; use serde::Deserialize; use serde_json::Value; use crate::{ CliResult, config::{Config, DEFAULT_RDR_BUFFER_CAPACITY, Delimiter}, util, }; #[derive(Deserialize)] struct Args { arg_input: Option, flag_output: Option, flag_delimiter: Option, flag_ignore_errors: bool, flag_jobs: Option, flag_batch: usize, } fn recurse_to_infer_headers(value: &Value, headers: &mut Vec>, path: &[String]) { match value { Value::Object(map) => { for (key, value) in map { match value { Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) | Value::Array(_) => { let mut full_path = path.to_owned(); full_path.push(key.to_string()); headers.push(full_path); }, Value::Object(_) => { let mut new_path = path.to_owned(); new_path.push(key.to_string()); recurse_to_infer_headers(value, headers, &new_path); }, #[allow(unreachable_patterns)] _ => {}, } } }, _ => { headers.push(vec![String::from("value")]); }, } } fn infer_headers(value: &Value) -> Vec> { let mut headers: Vec> = Vec::new(); recurse_to_infer_headers(value, &mut headers, &Vec::new()); headers } fn get_value_at_path(value: &Value, path: &[String]) -> Option { let mut current = value; for key in path { match current.get(key) { Some(new_value) => { current = new_value; }, None => { return None; }, } } Some(current.clone()) } #[inline] fn json_line_to_csv_record(value: &Value, headers: &[Vec]) -> csv::StringRecord { let mut record = csv::StringRecord::new(); for path in headers { let value = get_value_at_path(value, path); match value { Some(value) => { record.push_field(&match value { Value::Bool(v) => { if v { String::from("true") } else { String::from("false") } }, Value::Number(v) => v.to_string(), Value::String(v) => v, Value::Array(v) => v .iter() .map(std::string::ToString::to_string) .collect::>() .join(","), _ => String::new(), }); }, _ => { record.push_field(""); }, } } record } pub fn run(argv: &[&str]) -> CliResult<()> { let args: Args = util::get_args(USAGE, argv)?; let mut wtr = Config::new(args.flag_output.as_ref()) .delimiter(args.flag_delimiter) .writer()?; let mut is_stdin = false; let mut rdr: Box = match args.arg_input.clone() { None => { is_stdin = true; Box::new(BufReader::new(io::stdin())) }, Some(p) => Box::new(BufReader::with_capacity( DEFAULT_RDR_BUFFER_CAPACITY, fs::File::open(p)?, )), }; let mut headers: Vec> = Vec::new(); let mut headers_emitted: bool = false; // amortize memory allocation by reusing record let mut batch_line = String::new(); // reuse batch buffers let batchsize: usize = if args.flag_batch == 0 { if is_stdin { // if stdin, we don't know how many lines there are // so just make a reasonably big batch size 1_000_000 } else { // safety: we know arg_input is Some coz of the std_in check above util::count_lines_in_file(&args.arg_input.unwrap())? as usize } } else { args.flag_batch }; let mut batch = Vec::with_capacity(batchsize); let mut batch_results = Vec::with_capacity(batchsize); util::njobs(args.flag_jobs); let mut result_idx = 0_u64; 'batch_loop: loop { for _ in 0..batchsize { batch_line.clear(); match rdr.read_line(&mut batch_line) { Ok(0) => { // EOF break; }, Ok(_) => { batch.push(batch_line.clone()); }, Err(e) => { if args.flag_ignore_errors { continue; } return fail_clierror!( r#"Could not read input line!: {e} Use `--ignore-errors` option to skip malformed input lines. Use `tojsonl` command to convert _to_ jsonl instead of _from_ jsonl."#, ); }, } } if batch.is_empty() { break 'batch_loop; // EOF } if !headers_emitted { let value: Value = match serde_json::from_str(&batch[0]) { Ok(v) => v, Err(e) => { return fail_clierror!( "Could not parse first input line as JSON to infer headers: {e}", ); }, }; headers = infer_headers(&value); let headers_formatted = headers.iter().map(|v| v.join(".")).collect::>(); let headers_record = csv::StringRecord::from(headers_formatted); wtr.write_record(&headers_record)?; headers_emitted = true; } // do actual work via rayon batch .par_iter() .map(|json_line| match serde_json::from_str(json_line) { Ok(v) => Some(json_line_to_csv_record(&v, &headers)), Err(e) => { if !args.flag_ignore_errors { log::error!("serde_json::from_str error: {e:#?}"); } None }, }) .collect_into_vec(&mut batch_results); // rayon collect() guarantees original order, so we can just append results of each batch for result_record in &batch_results { result_idx += 1; if let Some(record) = result_record { wtr.write_record(record)?; } else if !args.flag_ignore_errors { // there was an error parsing a json line return fail_clierror!( r#"Could not parse input line {result_idx} as JSON Use `--ignore-errors` option to skip malformed input lines. Use `tojsonl` command to convert _to_ jsonl instead of _from_ jsonl."#, ); } } batch.clear(); } // end batch loop Ok(wtr.flush()?) }