On 1 billion row challenge
Crunching 14GB in 14 seconds through custom parsing, SIMD, and multi-threading.
rust systems-programming simd mmap algorithms datafusion
Introduction
Working with big data has always fascinated me. There's something compelling about watching systems transform massive amounts of raw data into meaningful insights. At first glance, data processing problems often appear deceptively simple, but the complexity emerges when you start optimizing for performance. The one billion row challenge (started quite a while back) is a perfect example of this: what seems like a straightforward aggregation task becomes an interesting exercise in systems programming once you try to make it fast.
Although I'm super late, I decided to tackle this challenge over a weekend. This post walks through my iterative approach to optimization and the lessons learned along the way.
Your mission, should you decide to accept it, is deceptively simple: write a Java program for retrieving temperature measurement values from a text file and calculating the min, mean, and max temperature per weather station. There’s just one caveat: the file has 1,000,000,000 rows!I'll go with rust as usual and start by solving this at a smaller scale (100 million rows) to iterate faster.
My system specs:
- M3 8 core (4P 4E)
- 24 GB LPDDR5 RAM
Since I work often with datafusion at work, I'm curious to see how it performs as a baseline benchmark.
How does DataFusion perform?
I'll use datafusion-cli as a baseline benchmark. The data is stored in a semicolon-delimited CSV file, so I'll create an external table with the appropriate configuration:
create external table measurements (
weather_station VARCHAR,
temperature FLOAT,
)
stored as csv
location './measurements.txt'
options ('has_header' 'false', 'delimiter' ';');Let's verify the table can read the CSV correctly:
> select * from measurements limit 5;
+-----------------+-------------+
| weather_station | temperature |
+-----------------+-------------+
| Islām Qal‘ah | 16.8 |
| Had Dra | -96.1 |
| Murlīganj | 88.7 |
| Ilfracombe | 59.7 |
| Warren | -42.1 |
+-----------------+-------------+
5 row(s) fetched.
Elapsed 0.015 seconds.
> select count(distinct weather_station) from measurements;
+----------------------------------------------+
| count(DISTINCT measurements.weather_station) |
+----------------------------------------------+
| 8906 |
+----------------------------------------------+
1 row(s) fetched.
Elapsed 0.154 seconds.Great! Now let's try to run the actual query for the challenge:
> select weather_station, min(temperature) as mint, max(temperature) as maxt, avg(temperature) as meant from measurements group by weather_station limit 5;
+-----------------+-------+------+---------------------+
| weather_station | mint | maxt | meant |
+-----------------+-------+------+---------------------+
| Darregueira | -99.5 | 99.7 | 0.8628336860429212 |
| Iesi | -99.8 | 99.9 | 1.287851877330262 |
| Galloway | -99.9 | 99.7 | 1.3548046681546106 |
| Bāhāgalpur | -98.7 | 99.7 | -1.912058907769182 |
| Bag‘dod | -99.8 | 99.9 | -0.7427704212895178 |
+-----------------+-------+------+---------------------+
5 row(s) fetched.
Elapsed 0.179 seconds.Well, it's quite fast on 10M rows. Let's check for 100M rows:
> select weather_station, min(temperature) as mint, max(temperature) as maxt, avg(temperature) as meant from measurements group by weather_station limit 5;
+-----------------+-------+------+----------------------+
| weather_station | mint | maxt | meant |
+-----------------+-------+------+----------------------+
| Somnāha | -99.9 | 99.9 | 0.2097603338817982 |
| Rocca di Papa | -99.9 | 99.9 | -0.4024348727702238 |
| Henderson | -99.9 | 99.9 | 0.16271185442686437 |
| Sinnamary | -99.9 | 99.9 | 0.04644437204311535 |
| Liushuquan | -99.9 | 99.9 | -0.20203168283785472 |
+-----------------+-------+------+----------------------+
5 row(s) fetched.
Elapsed 3.094 seconds.This gives us our baseline: 3.094 seconds for 100M rows.
How does DataFusion execute this query?
Running EXPLAIN reveals DataFusion's execution plan:
> explain select weather_station, min(temperature) as mint, max(temperature) as maxt, avg(temperature) as meant from measurements group by weather_station limit 5;
+---------------+-------------------------------+
| plan_type | plan |
+---------------+-------------------------------+
| physical_plan | ┌───────────────────────────┐ |
| | │ ProjectionExec │ |
| | │ -------------------- │ |
| | │ maxt: │ |
| | │ max(measurements │ |
| | │ .temperature) │ |
| | │ │ |
| | │ meant: │ |
| | │ avg(measurements │ |
| | │ .temperature) │ |
| | │ │ |
| | │ mint: │ |
| | │ min(measurements │ |
| | │ .temperature) │ |
| | │ │ |
| | │ weather_station: │ |
| | │ weather_station │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ CoalescePartitionsExec │ |
| | │ -------------------- │ |
| | │ limit: 5 │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ AggregateExec │ |
| | │ -------------------- │ |
| | │ aggr: │ |
| | │ min(measurements │ |
| | │ .temperature), │ |
| | │ max(measurements │ |
| | │ .temperature), avg │ |
| | │ (measurements.temperature)│ |
| | │ │ |
| | │ group_by: │ |
| | │ weather_station │ |
| | │ │ |
| | │ mode: │ |
| | │ FinalPartitioned │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ CoalesceBatchesExec │ |
| | │ -------------------- │ |
| | │ target_batch_size: │ |
| | │ 8192 │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ RepartitionExec │ |
| | │ -------------------- │ |
| | │ partition_count(in->out): │ |
| | │ 8 -> 8 │ |
| | │ │ |
| | │ partitioning_scheme: │ |
| | │ Hash([weather_station@0], │ |
| | │ 8) │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ AggregateExec │ |
| | │ -------------------- │ |
| | │ aggr: │ |
| | │ min(measurements │ |
| | │ .temperature), │ |
| | │ max(measurements │ |
| | │ .temperature), avg │ |
| | │ (measurements.temperature)│ |
| | │ │ |
| | │ group_by: │ |
| | │ weather_station │ |
| | │ │ |
| | │ mode: Partial │ |
| | └─────────────┬─────────────┘ |
| | ┌─────────────┴─────────────┐ |
| | │ DataSourceExec │ |
| | │ -------------------- │ |
| | │ files: 8 │ |
| | │ format: csv │ |
| | └───────────────────────────┘ |
| | |
+---------------+-------------------------------+
1 row(s) fetched.
Elapsed 0.007 seconds.Reading bottom-up, here's how DataFusion executes this query:
1. DataSourceExec: Splits the CSV file into 8 partitions (one per CPU core) to maximize parallelism
2. AggregateExec (Partial): Each partition computes partial aggregations independently
3. RepartitionExec: Redistributes data by weather station using hash partitioning
4. CoalesceBatchesExec: Combines smaller batches for more efficient vectorized operations
5. AggregateExec (Final): Merges partial results into final aggregations
6. CoalescePartitionsExec: Applies the LIMIT clause
7. ProjectionExec: Renames columns to their aliases
Implementation
Now let's build a custom solution that can match, or even beat, DataFusion's 3.094 second benchmark. I'll take an iterative approach, starting with a naive implementation and progressively optimizing based on profiling data.
Naive Implementation
The first version uses Rust's CSV reader to load the entire dataset into memory. While this isn't optimal, it gives us a baseline to measure against:
static WS_TEMP_MAP: Lazy<Vec<(String, f64)>> = Lazy::new(|| {
info!("Reading CSV file");
let mut list: Vec<(String, f64)> = Vec::with_capacity(1_000_000_000);
let mut reader = ReaderBuilder::new()
.delimiter(b';')
.from_path("./measurements_100M.txt")
.expect("CSV file missing");
let mut records = reader.records();
while let Some(Ok(record)) = records.next() {
let (weather_station, value) = (
record[0].to_owned(),
record[1].parse::<f64>().expect("invalid value"),
);
list.push((weather_station, value));
}
list
});Key points about this implementation:
- Pre-allocates a vector with 1B capacity to avoid reallocations (saving amortized allocation time)
- Loads ~1.4GB of data from disk into memory
- The CSV reader returns string arrays that are parsed into their respective types
The aggregation function groups temperatures by weather station and computes min, max, and average:
fn agg_values_per_ws(records: &[(String, f64)]) -> Vec<(String, f64, f64, f64)> {
let mut ws_temp_map: HashMap<String, Vec<f64>> = HashMap::new();
for (ws, value) in records {
ws_temp_map
.entry(ws.to_string())
.and_modify(|values| values.push(*value))
.or_insert(vec![*value]);
}
let values_per_ws: Vec<(String, f64, f64, f64)> = ws_temp_map
.into_iter()
.map(|(city, temps)| {
let min = temps.iter().copied().fold(f64::INFINITY, f64::min);
let max = temps.iter().copied().fold(f64::NEG_INFINITY, f64::max);
(
city,
(temps.iter().sum::<f64>() / temps.iter().len() as f64),
min,
max,
)
})
.collect();
values_per_ws
}
Result: 91 seconds for 100M rows
cargo run 93.12s user 1.96s system 97% cpu 1:37.51 totalThis is 30x slower than DataFusion! The flamegraph reveals the problem:
Almost 60% of the runtime is spent on reading the CSV. As expected, loading all of the data from disk to memory is not a good idea. Another thing to note is the time taken to clone the data is monstrous. We have two clear action items for now:
- Make an CSV Reader iterator. This will load the data onto memory as needed.
- Reduce the clones for the data read.
CSV parsing and using ByteRecords
After working on the above points, the function now looks like this.
fn get_csv_reader(path: PathBuf) -> csv::Reader<std::fs::File> {
info!("Reading CSV file");
ReaderBuilder::new()
.delimiter(b';')
.from_path(&path)
.expect("CSV file missing")
}
fn agg_values_per_ws(mut reader: csv::Reader<std::fs::File>) -> Vec<(String, f32, f32, f32)> {
let mut records_iter = reader.byte_records();
let mut ws_temp_map: HashMap<String, Vec<f32>> = HashMap::with_capacity(9000);
while let Some(Ok(record)) = records_iter.next() {
let mut record_iter = record.iter();
let ws_ref = unsafe { str::from_utf8_unchecked(record_iter.next().unwrap()) };
let temp_str = unsafe { str::from_utf8_unchecked(record_iter.next().unwrap()) };
let temp_ref = temp_str.parse::<f32>().unwrap();
if let Some(values) = ws_temp_map.get_mut(ws_ref) {
values.push(temp_ref);
} else {
ws_temp_map.insert(ws_ref.to_string(), vec![temp_ref]);
}
}
let values_per_ws: Vec<(String, f32, f32, f32)> = ws_temp_map
.into_iter()
.map(|(city, temps)| {
let min = temps.iter().copied().fold(f32::INFINITY, f32::min);
let max = temps.iter().copied().fold(f32::NEG_INFINITY, f32::max);
(
city,
(temps.iter().sum::<f32>() / temps.iter().len() as f32),
min,
max,
)
})
.collect();
values_per_ws
}Now we no longer have a global map for the data, rather we created an iterator which reduces the amount of memory used for loading the csv to memory.
Using unsafe here is actually okay because I know the data is not corrupted. This will skip utf8 validation which seems a bit unnecessary.
Benchmarking this version, showed that we shaved off almost 15 seconds.
cargo run -- --path measurements_100M.txt 71.38s user 0.59s system 98% cpu 1:13.27 totalIt was time to write our own parser, and so I did.
// aggs_values_per_ws function
let mut buf = Vec::with_capacity(1000);
while reader.read_until(b'\n', &mut buf).is_ok_and(|n| n > 0) {
let mut buf_chunks = buf.splitn(2, |b| b == &b';');
if let (Some(ws_bytes), Some(temp_bytes)) = (buf_chunks.next(), buf_chunks.next()) {
// ignore the `\n` at the end
let temp_bytes = temp_bytes.trim_ascii_end();
let ws_str = unsafe { str::from_utf8_unchecked(ws_bytes) };
let temp_ref = unsafe { str::from_utf8_unchecked(temp_bytes).parse::<f32>().unwrap() };
if let Some(values) = ws_temp_map.get_mut(ws_str) {
values.push(temp_ref);
} else {
ws_temp_map.insert(ws_str.to_string(), vec![temp_ref]);
}
}
buf.clear();
}The next version shaved off almost 23 seconds off the reader time.
cargo run -- --path measurements_100M.txt 50.28s user 0.49s system 98% cpu 51.410 total51 seconds for a non-release binary was not bad. I ran this version in release mode (finally! ;P) and it took 10 seconds. Nice, but we want to reach the sub-3s mark. Sampling the data again, one thing was clear to me that we are closing to a point where we have to use multi-threading, but something else took me by surprise.
Split operations took 15% of execution time. This was intriguing, so I did some digging and found splits add an overhead of validating the data is utf8 and you can see this when splitting the string, it returns a slice of the string itself. We should be faster by just using bytes here and converting bytes.
Parting ways with Strings
So after struggling with the compiler to strive for readable yet performant code, I came up with this:
fn agg_values_per_ws(mut reader: BufReader<std::fs::File>) -> Vec<(String, f32, f32, f32)> {
let mut ws_temp_map: HashMap<String, Vec<f32>> = HashMap::with_capacity(9000);
let mut buf = Vec::with_capacity(1000);
while reader.read_until(b'\n', &mut buf).is_ok_and(|n| n > 0) {
// ignore the `\n`
let mut buf_chunks = buf.splitn(2, |b| b == &b';');
if let (Some(ws_bytes), Some(temp_bytes)) = (buf_chunks.next(), buf_chunks.next()) {
let temp_bytes = temp_bytes.trim_ascii_end();
let ws_str = unsafe { str::from_utf8_unchecked(ws_bytes) };
let temp_ref = unsafe { str::from_utf8_unchecked(temp_bytes).parse::<f32>().unwrap() };
if let Some(values) = ws_temp_map.get_mut(ws_str) {
values.push(temp_ref);
} else {
ws_temp_map.insert(ws_str.to_string(), vec![temp_ref]);
}
}
buf.clear();
}
let values_per_ws: Vec<(String, f32, f32, f32)> = ws_temp_map
.into_iter()
.map(|(city, temps)| {
let (mut min, mut max, mut sum, len) =
(f32::INFINITY, f32::NEG_INFINITY, 0.0, temps.len());
for temp in temps {
min = f32::min(min, temp);
max = f32::max(max, temp);
sum += temp;
}
(city, sum / len as f32, min, max)
})
.collect();
values_per_ws
}This shaved off 3 more seconds and brought the execution down to
cargo run --release -- --path measurements_100M.txt 4.96s user 0.39s system 79% cpu 6.703 totalRefactor 1
I tried to improve values_per_ws logic to work on a single pass rather than multiple passes but it did not have significant impact. I think it must be due to compiler optimization the previous code was translated to same assembly instructions where it did not matter if you did a single pass or multiple passes. I will explore further into this territory in some other blog.
Asyncification
It was time that I try parallelism for once and see the actual benefits. The plan is to distribute the workload across threads depending on the number of CPUs available, each task will work on a subset of code and then return the final results to the main thread.
The problem is distribution. How to divide the dataset between threads such that each thread has equal data to process? First solution that came to my mind was to get the file size from metadata, divide it by the number of CPUs and Voila!!... Sadly, there is a big issue with this. The data may not be split correctly! It may split Mumbai;14.5 into Mum and bai;14.5!
To tackle this, I added a file chunking step which ensures that we do not corrupt any segment of the file. The function simply divides the workload into cpu_num segments and makes sure no line is split in between.
fn chunk_file(mut reader: BufReader<std::fs::File>, file_size: u64) -> Vec<(u64, u64)> {
let cpu_count = std::thread::available_parallelism().unwrap().get();
let logical_chunk_len = file_size / cpu_count as u64;
let mut offset = 0u64;
(0..cpu_count)
.map(|_| {
let start = offset;
offset += logical_chunk_len;
if offset > file_size {
// last segment
(start, file_size - 1)
} else {
let mut buf = vec![];
reader.seek_relative(logical_chunk_len as i64).unwrap();
let correction = reader.read_until(b'\n', &mut buf).unwrap();
offset += correction as u64;
(start, offset)
}
})
.collect()
}
For parallelism, I could have introduced a mutual exclusion (note: In this scenario the most suitable lock is Mutex, and not RwLock because this workload is write heavy) between workers but did not use it because I could see it being done without it. Each segment can calculate its own summary for their respective chunk. Summary is a primitive aggregation of values in the chunk. Primitive because mean is broken down to sum and len.
#[derive(Debug, Clone)]
struct WeatherStationSummary {
min: f32,
max: f32,
sum: f32,
len: u64,
}
impl WeatherStationSummary {
pub fn new(temp: f32) -> WeatherStationSummary {
WeatherStationSummary {
min: temp,
max: temp,
sum: temp,
len: 1,
}
}
pub fn record_temp(&mut self, temp: f32) {
self.min = f32::min(self.min, temp);
self.max = f32::max(self.max, temp);
self.sum += temp;
self.len += 1;
}
pub fn merge(&mut self, other: &Self) {
self.min = f32::min(self.min, other.min);
self.max = f32::max(self.max, other.max);
self.sum += other.sum;
self.len += other.len;
}
}
impl Display for WeatherStationSummary {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"min: {}, max: {}, avg: {}",
self.min,
self.max,
self.sum / self.len as f32
)
}
}Finally, for the aggregation step, the function is mainly split into two steps.
1. Spawning worker threads for each segment
2. Merging results from each worker into final map
fn agg_values_per_ws(
path: PathBuf,
chunks: &[(u64, u64)],
) -> HashMap<String, WeatherStationSummary> {
std::thread::scope(|s| {
let handles: Vec<ScopedJoinHandle<HashMap<String, WeatherStationSummary>>> = chunks
.iter()
.map(|(start, end)| {
s.spawn(|| {
let (mut reader, _) = get_csv_reader(&path);
reader.seek(std::io::SeekFrom::Start(*start)).unwrap();
let mut ws_temp_map: HashMap<String, WeatherStationSummary> = HashMap::new();
let mut offset = *start;
let mut buf = Vec::with_capacity(1000);
while let Ok(n) = reader.read_until(b'\n', &mut buf) {
if offset + n as u64 > *end {
break;
}
offset += n as u64;
let mut buf_chunks = buf.splitn(2, |b| b == &b';');
if let (Some(ws_bytes), Some(temp_bytes)) =
(buf_chunks.next(), buf_chunks.next())
{
// ignore the `\n` at the end
let temp_bytes = temp_bytes.trim_ascii_end();
let ws_str = unsafe { str::from_utf8_unchecked(ws_bytes) };
let temp = unsafe {
str::from_utf8_unchecked(temp_bytes).parse::<f32>().unwrap()
};
if let Some(summary) = ws_temp_map.get_mut(ws_str) {
summary.record_temp(temp);
} else {
ws_temp_map
.insert(ws_str.to_string(), WeatherStationSummary::new(temp));
}
}
buf.clear();
}
ws_temp_map
})
})
.collect();
handles.into_iter().fold(
HashMap::new(),
|mut final_map: HashMap<String, WeatherStationSummary>, handle| {
if let Ok(map) = handle.join() {
for (name, summary) in map {
final_map
.entry(name)
.and_modify(|old_summary| old_summary.merge(&summary))
.or_insert(summary);
}
}
final_map
},
)
})
}Now to benchmarking, this improved the execution times to 4 seconds range. It was a little disappointing because I thought this would for sure bring it down to <3 seconds range.
cargo run --release -- --path measurements_100M.txt 4.98s user 0.66s system 138% cpu 4.083 totalFlamegraph showed that we spend most of the time reading:
That was kind of disappointing at first, I expected to go under 3 seconds of execution time with threads but then I looked at the time. I saw that memchr is where 16% of CPU cycles went and 12% on HashMap. Since I have an idea to improve the hashmap logic, I proceeded with that for now.
Faster Hashmap
Basically we need a faster hashing function. After replacing std Hashmap with FxHashmap and saw a drop of 1 second from the runtime.
cargo run --release -- --path measurements_100M.txt 4.38s user 0.52s system 144% cpu 3.395 totalUpon seeing the flamegraph, I realized splitn is not really an efficient way to split the line. I decided to try memchr as it fits perfectly for this case.
Memchr ( SIMD )
cargo run --release -- --path measurements_100M.txt 4.21s user 0.25s system 233% cpu 1.909 totalAs expected, You rarely lose, but often gain big.This optimization could be attributed to SIMD instructions for string manipulation. At this point we already surpassed our goal of beating datafusion.
Finally let's try to crunch 1 Billion rows
I generated the dataset from 1brc repo, it was about 14.3 GB. I was hoping to land somewhere between 20-30 seconds range because if 100M took 2 seconds, 1B will take 20 seconds. And surely the results were aligned:
cargo run --release -- --path measurements_1B.txt 43.87s user 9.58s system 268% cpu 19.903 totalMMAP
However, this made me think whether I should try reading this huge file via MMAP. By bypassing the user space buffer, maybe we get better speed. But there was no improvement. I think the reason behind this is the type of workload we have here. It's because of a lot of sequential reads and not random access.
I reverted all the changes for mmap, because on top of providing no benefit, the code just looked uglier.
More threads
I took a look at my flamegraph again and saw a wide wall of read syscall. Like the Night king facing the wall in the north (south for him), I wanted to bring it down. I hoped to do this with MMAP, but it failed.
But there was a sign. An observant reader must have noticed that the time results of our execution show system % never goes above 250%. This means we are using 2.3 CPU cores only and not CPU bound at this point but IO bound. Generally, if your program is doing a lot of IO and is not CPU bound, it's advisable to try more threads to do more IO in parallel.
I refactored the code a bit and added a cli argument to set the number of threads
cargo run --release -- --path measurements_1B.txt -n 20 73.94s user 22.97s system 680% cpu 14.250 totalI noticed If I have anywhere between 16-20 threads. It will use almost 700% CPU. That means we moved towards being CPU bound. Awesome!
Final version
Here are the main functions for my program after all the iterations:
fn generate_chunks(
mut reader: BufReader<std::fs::File>,
file_size: u64,
tc: Option<usize>,
) -> Vec<(u64, u64)> {
let cpu_count = if let Some(thread_count) = tc {
thread_count
} else {
std::thread::available_parallelism().unwrap().get()
};
log::info!("Using {cpu_count} threads");
let logical_chunk_len = file_size / cpu_count as u64;
let mut offset = 0u64;
(0..cpu_count)
.map(|_| {
let start = offset;
offset += logical_chunk_len;
if offset > file_size {
// last segment
(start, file_size - 1)
} else {
let mut buf = vec![];
reader.seek_relative(logical_chunk_len as i64).unwrap();
let correction = reader.read_until(b'\n', &mut buf).unwrap();
offset += correction as u64;
(start, offset)
}
})
.collect()
}
fn process_chunk(
mut reader: BufReader<File>,
start: u64,
end: u64,
) -> FxHashMap<String, WeatherStationSummary> {
reader.seek(std::io::SeekFrom::Start(start)).unwrap();
let mut ws_temp_map: FxHashMap<String, WeatherStationSummary> =
FxHashMap::with_capacity_and_hasher(1024, FxBuildHasher::default());
let mut offset = start;
let mut buf = Vec::with_capacity(2048);
while let Ok(n) = reader.read_until(b'\n', &mut buf) {
if offset + n as u64 > end {
break;
}
offset += n as u64;
if let Some(pos) = memchr(b';', &buf) {
let ws_bytes = &buf[..pos];
// ignore the `\n` at the end
let temp_bytes = buf[pos + 1..].trim_ascii_end();
let ws_str = unsafe { str::from_utf8_unchecked(ws_bytes) };
let temp = unsafe { str::from_utf8_unchecked(temp_bytes).parse::<f32>().unwrap() };
if let Some(summary) = ws_temp_map.get_mut(ws_str) {
summary.record_temp(temp);
} else {
ws_temp_map.insert(ws_str.to_string(), WeatherStationSummary::new(temp));
}
}
buf.clear();
}
ws_temp_map
}
fn agg_values_per_ws(
path: PathBuf,
chunks: &[(u64, u64)],
) -> FxHashMap<String, WeatherStationSummary> {
std::thread::scope(|s| {
let handles: Vec<ScopedJoinHandle<FxHashMap<String, WeatherStationSummary>>> = chunks
.iter()
.map(|(start, end)| {
s.spawn(|| {
let file = File::open(&path).unwrap();
let reader = BufReader::new(file);
process_chunk(reader, *start, *end)
})
})
.collect();
handles.into_iter().fold(
FxHashMap::with_capacity_and_hasher(1024, FxBuildHasher::default()),
|mut final_map: FxHashMap<String, WeatherStationSummary>, handle| {
if let Ok(map) = handle.join() {
for (name, summary) in map {
final_map
.entry(name)
.and_modify(|old_summary| old_summary.merge(&summary))
.or_insert(summary);
}
}
final_map
},
)
})
}I think with the current solution we are at a good place. I'm sure there are more corrections I can do to improve this further.
I tried jemalloc as I thought arena buffers for each thread might improve the allocation time. Unsurprisingly, it added no benefit. I could have predicted this just by looking at the flamegraph but I wanted to give it a shot nonetheless.
That's it for this one, folks. If you are interested, I'll be posting some more ideas and updates on my twitter. Feel free to reach out to talk more on this or systems in general! Here is the final code: 1brc
References
We learn by reading, experimenting and sharing. On this journey, I came across some well curated articles:
- <https://pncnmnp.github.io/blogs/one-billion-row-challenge.html>
- <https://curiouscoding.nl/posts/1brc/>
- Burntsushi has some great articles on CSV parsing and has a whole crate about this
rust-csv - Datafusion is a great library which you should definitely check out if you are into databases, data pipelines or work with lot of data in general.