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] [<input>]
qsv jsonl --help
jsonl options:
--ignore-errors Skip malformed input lines.
-j, --jobs <arg> 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 <size> 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 <file> Write output to <file> instead of stdout.
-d, --delimiter <arg> 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<String>,
flag_output: Option<String>,
flag_delimiter: Option<Delimiter>,
flag_ignore_errors: bool,
flag_jobs: Option<usize>,
flag_batch: usize,
}
fn recurse_to_infer_headers(value: &Value, headers: &mut Vec<Vec<String>>, 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<Vec<String>> {
let mut headers: Vec<Vec<String>> = Vec::new();
recurse_to_infer_headers(value, &mut headers, &Vec::new());
headers
}
fn get_value_at_path(value: &Value, path: &[String]) -> Option<Value> {
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<String>]) -> 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::<Vec<_>>()
.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<dyn BufRead> = 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<String>> = Vec::new();
let mut headers_emitted: bool = false;
let mut batch_line = String::new();
let batchsize: usize = if args.flag_batch == 0 {
if is_stdin {
1_000_000
} else {
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) => {
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; }
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::<Vec<String>>();
let headers_record = csv::StringRecord::from(headers_formatted);
wtr.write_record(&headers_record)?;
headers_emitted = true;
}
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);
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 {
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();
}
Ok(wtr.flush()?)
}