Repository navigation
Parallel NDSON file reading #8502
Description
Is your feature request related to a problem or challenge?
DataFusion can now automatically read CSV and parquet files in parallel (see #6325 for CSV)
It would be great to do the same for "NDJSON" files -- namely files that have multiple JSON objects placed one after the other.
Describe the solution you'd like
Basically implement what is described in #6325 for JSON -- and read a single large ND json file (new line delimited file) in parallel
Describe alternatives you've considered
Some research may be required -- I am not sure if finding record boundaries is feasible
Additional context
I found this while writing tests for #8451
Activity
[-]Parallel NDSON reading[/-][+]Parallel NDSON file reading[/+]tustvold commentedon Dec 12, 2023
This should be simpler than CSV, as NDJSON does not typically permit unescaped newline characters, so it should just be a case of finding the next newline
alamb commentedon Dec 12, 2023
I think this is a medium difficulty task for a new contributor as the pattern exists and there are tests (e.g. see #8505)
JacobOgle commentedon Dec 13, 2023
@alamb I wouldn't mind digging into this one if its still open
alamb commentedon Dec 13, 2023
I just filed it and I don't know of anyone else working on it.
Thanks @JacobOgle
kassoudkt commentedon Dec 19, 2023
Is it still available ? i wloud love to take it :)
JacobOgle commentedon Dec 19, 2023
@kassoudkt feel free! I've been a bit tied up lately so if you're free go for it!
marvinlanhenke commentedon Dec 23, 2023
@tustvold @alamb
...out of curiosity, I was digging into this as well. From my understanding (looking at the CSV impl) the
FileGroupPartitionerand its methodrepartition_file_groupsare used to create the partitions. However, in this case evenly divided by size.In order for NDJSON to be split "correctly" (and not in the middle of a JSON Object) the FileGroupPartitioner needs a new method to split on newline? Would this be a reasonable approach?
Then only
fn repartitionedof trait ExecutionPlan andfn openof trait FileOpener need to be implemented.Thanks for helping out.
tustvold commentedon Dec 23, 2023
The way it typically works is the split is based on filesize but the reader is setup such that one of the bounds includes the current partial row, and the other does not. For example the reader starts at the NEXT newline (with special case for first row) and stops when it reaches the end of a line AND the byte position now exceeds the end limit. CSV (and parquet) behave similarly.
This all avoids the planner needing to perform IO, which is pretty important
marvinlanhenke commentedon Dec 23, 2023
I just realized that I forgot the IO part. Now, I understand the approach better - thanks for the explanation.
alamb commentedon Dec 24, 2023
Indeed -- I think the relevant code that finds the next bounds is https://github.com/apache/arrow-datafusion/blob/a1e959d87a66da7060bd005b1993b824c0683a63/datafusion/core/src/datasource/physical_plan/csv.rs#L411-L450
alamb commentedon Dec 24, 2023
BTW this comment might help: https://github.com/apache/arrow-datafusion/blob/6b433a839948c406a41128186e81572ec1fff689/datafusion/core/src/datasource/physical_plan/file_groups.rs#L35-L79
marvinlanhenke commentedon Dec 25, 2023
@alamb thanks for the pointers.
I already implemented a working solution, however I need to do some refactoring (if my kids let me :P ).
I'd also like to extract the common functionality since the NdJson and the CSV implementation are nearly the same; any suggestions where to put those
utilityfunctions, likefind_first_newline? I think mod.rs would be fine.alamb commentedon Dec 26, 2023
arrow-datafusion/datafusion/core/src/datasource/physical_plan/mod.rs sounds like a good idea to me
marvinlanhenke commentedon Dec 26, 2023
@alamb
...found some time to clean up the changes.
However, I am not sure about properly benchmarking the solution (as stated in the PR) and perhaps some more tests are needed? I am looking forward to your feedback.
alamb commentedon Dec 27, 2023
I think a good test would be to find a largish JSON input file and show some benchmark reading numbers
I don't know of any existing benchmarks we have for reading large JSON files. Maybe we could add a benchmark for reading from large JSON (any CSV?) files in https://github.com/apache/arrow-datafusion/tree/main/benchmarks#datafusion-benchmarks
Something like
That would measure the speed of parsing a large JSON file
marvinlanhenke commentedon Dec 27, 2023
@alamb
I did some basic benchmarking.
Methodology:
create external table json_test stored as json location '/home/ml/data_60m.json';select * from json_test;&select * from json_test where a > 5;Results:
select * from json_test;select * from json_test where a > 5;When applying a filter and
explain select * from json_test where a > 5;we can see the repartitioning happening (file_groups: 12).
However, when simply running
select * from json_test.File_groups remain at 1 and we get no parallel reading.
I think this issue relates to: #6983
Haven't tested it with a dataframe; however the issue seems to remain, at least for the datafusion-cli
(tested with JSON and CSV)
alamb commentedon Dec 28, 2023
A good test of just the speed of the parsing might be something like
That will minimize most of the actual query work other than parsing and won't need to try and format / carry through the other columns
marvinlanhenke commentedon Dec 28, 2023
...the updated result with different queries:
select count(*) from json_test;select count(*) from json_test where a > 5;Closes #8502: Parallel NDJSON file reading (#8659)