Hacker News Reader

Comment by jandrewrogers | Hacker News Reader

Many developers severely underestimate how much workload can be served by a single modern server and high-quality C++ systems code. I've scaled distributed workloads 10x by moving them to a single server and a different software architecture more suited for scale-up, dramatically reducing system complexity as a bonus. The number of compute workloads I see that actually need scale-out is vanishingly small even in industries known for their data intensity. You can often serve millions of requests per second from a single server even when most of your data model resides on disk.

We've become so accustomed to extremely inefficient software systems that we've lost all perspective on what is possible.

View on HN

Can you expand on this? I have some pretty massive compute loads that need to be scaled onto a cluster with 100+ workers for most computations. This is after I use a library called dask that graphically does its own mapreduce optimisation inside its modules. This is all for a relatively small 250GB raw data file that I keep in a csv (and need to convert to SQL at some point).

Are you saying this can be optimised to fit inside a single 10 core server in terms of compute loads?

Don't know why you're being downvoted but I'll assume your question is genuine.

You use a cluster when your data and compute requirements are large and parallel enough that the tax paid on network latency trumps the 10-20X speedup you get on SSD and 1000X speedup you get from just keeping data in RAM.

250 Gigs is tiny enough that you could probably much get better performance running on high memory instance in AWS or GCP. You'll generally have to write your own multiprocessing code though which is fairly simple - your existing library may also be able to support it.

I once actually ran this kind of workload on just my laptop using a compiled language that performed better than pyspark on a cluster.

I'd love to keep it in RAM if I could. The problem is, the library I'm familiar with (pandas) typically seems to take more memory than the original csv file once it loads it onto memory. I know this is due to bad data types but in certain cases, I cannot get around those.

However, even if I could load it all into memory at once, and assuming it takes 200 gb, I'm still using a master's student access to a cluster. So I get preempted like it's nobody's business. Hence why I prefer a smaller memory footprint even if I take up cpus at variable rates through a single execution.

I did try to write my own multiprocessing code for this, but the operations are sometimes too complicated (like groupby) for me to rewrite everything from the ground up. If I'm not reliant on serial data communication between processes (like you'd need to sort a column), I can get it done pretty easily. In fact, I wrote my data cleaning code with this and cleaned up the entire file in half an hour because single chunks didn't rely on others.

However, if you have some idea of how to run these computational loads in parallel in python or any other language on single compute instances (like the size of a laptop's memory of 16 gb), I'd really love to see it. Thanks.

Numpy supports memory mapping `ndarrays` which can back a DataFrame in pandas. This lets you access a dataset far larger than will fit in RAM as if it lived in RAM. Provided it's on fast SSD storage you'll have speedy access to the data and can process huge chunks at once.

Can you provide a link to this please? My current knowledge is that all numpy data lives in memory, and pandas itself has a feature to fragment any data into iterables so I can read upto my memory limit. I cannot use this feature due to the serial nature of some of the operations that I alluded to (I'd have to almost rewrite the entire library for some of these complicated operations like groupby and sorting).

I do have fast SSD storage because it's on the scratch drive of a cluster and from what I've seen it can do ~300-400 MB/s easily. I haven't had a chance to test more than that since I'm mostly memory constrained in much of my testing.

My current attempt is to push this data into a pure database handling system like SQL so that I can query it. But like I said, I work with a less-than-stellar set of tools and I have to literally set up a postgres server from ground up to write to it. Which shouldn't be a big deal except when it's on a non-root user and I have to keep remapping dependencies (took 5-6 hours to set it up on the instance I have access to).

My other option was to write the entire 250 GB to a sqllite database using the sqlalchemy library in Python, but that seems to fail whether I do it with parallel writes, or serial writes. In both cases, it fails after I create ~64-70 tables.

Dask groupby example: https://examples.dask.org/dataframes/02-groupby.html

> Generally speaking, Dask.dataframe groupby-aggregations are roughly same performance as Pandas groupby-aggregations, just more scalable.

The dask.distributed scheduler can also run on one high-RAM instance (with threads or processes) https://docs.dask.org/en/latest/setup.html

Pandas docs > Ecosystem > Out-of-core: https://pandas.pydata.org/pandas-docs/stable/ecosystem.html#...

Reading from Parquet into Apache Arrow is much faster than CSV because the data can just be directly loaded into RAM. https://ursalabs.org/blog/2019-10-columnar-perf/

If you have GPU instances, cuDF has a Pandas-like API on top of Apache Arrow. https://github.com/rapidsai/cudf

> Built based on the Apache Arrow columnar memory format, cuDF is a GPU DataFrame library for loading, joining, aggregating, filtering, and otherwise manipulating data.

> cuDF provides a pandas-like API that will be familiar to data engineers & data scientists, so they can use it to easily accelerate their workflows without going into the details of CUDA programming.

Dask-ML makes scalable scikit-learn, XGBoost, TensorFlow really easy. https://dask-ml.readthedocs.io/en/latest/

... re: the OT: While it's possible to write C++ code that's really fast, it's generally inflexible, expensive to develop, and dangerous for devs with experience in their respective domains of experience to write. Much saner to put a Python API on top and optimize that during compilation.

There are a few C++ frameworks in the top quartile of the TechEmpower framework benchmarks. https://www.techempower.com/benchmarks/

Hardware/hosting is relatively cheap. Developers and memory vulnerabilities aren't.

Unfortunately, I'm not sure what's wrong with dask, but it doesn't work properly on my cluster. I tested it on an exceedingly simple operation - find all unique values in a very big column (5 billion rows, but I know for a fact that there are only 500-502 unique values in there). With a 100 workers, it still failed. Now this is an embarrassingly parallel operation that can be implemented trivially. So I'm not sure if there's a problem with my cluster or if dask just does not work with slurm clusters very well.

https://docs.dask.org/en/latest/setup/hpc.html says dask-jobqueue handles "PBS, SLURM, LSF, SGE and other resource managers"

"Dask on HPC, what works and what doesn't" https://github.com/dask/dask-blog/issues/5

Maybe you should spend some time developing a job visualization system for end users from scratch, for end users with lots of C, JS, and HTML experience https://jobqueue.dask.org/en/latest/interactive.html

I'd spin up something like AWS r5.16xlarge node for the processing just for this and shut it down after use - should cost few 10s of dollars per run or so. Of course in some corporate environments, this option may not be available to you.

https://docs.scipy.org/doc/numpy/reference/generated/numpy.m...

You can create memory mapped ndarrays, these act like normal numpy arrays but don't need to fit into RAM. Numpy maps the array to a binary file on disk. The array otherwise acts like an ndarray so you can build a DataFrame with it. Whenever you access an array index Numpy in the background (essentially) seeks that many values into the file to grab the value of that index.

Since you're on a fast SSD and Numpy is fairly smart you'll be able to access your arrays close to your drive's speed. It's slower than if the whole database was in RAM but far faster than distributing the data over a network to a bunch of worker nodes. Memory mapped files let you have array-like access to data on disk as if it lived in RAM. When building a pandas DataFrame from a memmapped ndarray I believe you just need to set copy=False in the constructor for it to Just Work.

I don't know what your data looks like but I doubt loading it into SQLite is going to improve your performance.

Unfortunately my data isn't all numbers. It has text too. The sparse examples in that link only show this for reading in numbers. Do you know off hand if it translates well? There is a dtype parameter, but it'll take me a few days to get back to this code, so I figured I'd check beforehand.

Your second paragraph is essentially what I want. I'm willing to wait a day for code that may run in 1 hour from memory, so time isn't entirely an issue unless it's starting to bleed into weeks. The read_csv function in pandas has a parameter called memory_map, but when I tried using it on a smaller 7GB dataset, it read the whole thing into memory (32GB instance) even when I set it to True.

SQLite is definitely not my best option here. It was the only server-less implementation I could find, so I tried to use it and it didn't work. However, a database like implementation will be helpful because each operation I need to do will require data that satisfies certain timestamp and arithmetic conditions. I figured it'd be best to load the whole thing into a db and query it for every operation to train my model.

> You use a cluster when your data and compute requirements are large and parallel enough that the tax paid on network latency trumps the 10-20X speedup you get on SSD and 1000X speedup you get from just keeping data in RAM.

Nowadays there's another reason to use clusters: to autoscale your expenditure wrt workload. A little inefficiency might be acceptable if you don't have to pay for a huge beefed up server idling at, say, 30%.

note: doing anything on a 250gb file in python will require a lot more ram than 250gb. generally my expectation is I will need 10x the ram as the size of the file when using pandas, for when I accidentally do something that triggers pathological behavior.

I'm doing my PhD in HPC (part of my work is in situ stuff). One of the biggest problems is actually IO. But honestly, 250GB isn't that big. I haven't used dask, but I know people that do. I just avoid python for anything I need performance for, C++ is just always going to be faster.

Biased: you might want to look into DOE libraries. For IO I suggest ADIOS2 [0]. There's python bindings too.

One of the biggest things you can do is use a different storage type like BP (ADIOS) or hdf5. These are readable but binary. But to really determine how to speed up your problem you have to know where the bottleneck is. Is it IO or compute? With 100 workers (threads or nodes?) you aren't highly parallelized. I mean that could be a single node if it's threads.

[0] https://github.com/ornladios/ADIOS2

Not who you asked, but, it's hard to say without knowing exactly what computation is being done, or how much of the time is spent on IO. If you organize that 250GB in ram the right way (cache coherency, right container types), and spend a lot of effort doing analysis of algorithm selection, you might be surprised how much you can get done on a single (large) machine with enough cores.

Cache locality, not coherency.

I'm curious as to why this seems like a common mistake (I've seen it a few times already in this comment section).

Not sure which DB you are using, but you can load the csv file into the DB directly on a single thread using something like LOAD DATA INFILE.

If you have some good indexes and do some push-down work (give the database aggregation tasks to do instead of your python code), you should probably be more than fine.

For a 250Gb file.. should be ok.. maybe add some partitioning too.

I'm open to using any db that I can query over some engine with a python implementation. So any SQL db should be fine. However, I don't know how to convert a csv to an SQL directly. Is the command you mentioned part of some SQL server package? Sounds like it's exactly what I need.

pandas can read from a CSV file and then write to SQL. Even if you don't go the SQL route, you'd probably gain significant benefits by working with HDF instead of CSV.

https://pandas.pydata.org/pandas-docs/version/0.22/generated...

https://pandas.pydata.org/pandas-docs/version/0.22/io.html#w...

250GB data is tiny. There are laptops with 128GB RAM, let alone servers.

Obligatory read: https://aadrake.com/command-line-tools-can-be-235x-faster-th...

Yes, but I'm heavily constrained because my access to the cluster I'm using is very low level and I get pre-empted quite a bit in my tasks. I'm probably over stretching between the amount of data I need to handle and my severe lack of skills (I'm a quant in training).

> Are you saying this can be optimised to fit inside a single 10 core server in terms of compute loads?

I'm currently employed to write software which does DNA analysis. DNA is known for being Big Data.

Some applications are very compute intensive and others are very data-relation intensive. The very compute intensive applications process about 1GB of data in about 30 minutes on a 32 core Xeon 6xxx with 32GB of RAM assigned to it. The data-relation intensive application processes about 350GB of data in 15 minutes on a single core of the same CPU but with 700GB of RAM assigned to it.

Both are heavily optimized but in different ways. So without knowing more about what you're doing with that data, it's hard to say.

That's very interesting. Do you mind sharing some more details on what those computations are? Where can I learn more about these DNA analysis use cases?

I've found that Wikipedia has really good high-level information for a lot of the subject matter. For use cases, the business sells direct-to-consumer DNA tests. I've worked on several of the analysis products' softwares.

The compute-heavy workload calculates edit distance [0] of short paired-end sequenced DNA [1] vs the human genome [2]. There is open source software to manipulate the FASTA/FASTQ [3] and SAM files [4] and run the calculations [5]. The aligned file is processed in a couple of minutes to genetic variation report [6] which is used for some of the analysis products that were purchased. One popular product will give you a haplogroup [7] which basically tells you where you are in a genetic tree.

The relationship estimator uses a different sequencing technology and basically consumes a CSV file from the sequencer's manufacturer. It uses a proprietary algorithm to calculate centimorgans [8]. That then gives relationship estimates between you and other people who've purchased the product.

[0]: https://en.wikipedia.org/wiki/Needleman%E2%80%93Wunsch_algor...

[1]: https://en.wikipedia.org/wiki/Next-generation_sequencing

[2]: https://en.wikipedia.org/wiki/Reference_genome

[3]: https://en.wikipedia.org/wiki/FASTQ_format

[4]: https://github.com/samtools/hts-specs

[5]: https://en.wikipedia.org/wiki/List_of_sequence_alignment_sof...

[6]: https://en.wikipedia.org/wiki/Variant_Call_Format

[7]: https://en.wikipedia.org/wiki/Haplogroup

[8]: https://en.wikipedia.org/wiki/Centimorgan

So my entire dataset is ~24 x 250GB files. That 24 number can be larger if I can find an efficient way of processing each 250GB chunk. Each 250GB chunk is actually stock tick data so it has 500 stocks inside it. A heavily traded stock takes up ~10 GB of memory while a very thinly traded stock can top out at just 700 MB.

While I hope that each 250GB chunk has everything in order and I can separate it cleanly, I don't trust it. So I broke up all the files by stock in an embarrassingly parallelised code I wrote and got 500 separate files. However, the problem is, I'd want to process these 500 files parallely so each stock gets only so much memory (and hence, my original constraint remains). And for each stock, I want the data to be queried on timestamps, so I need a way to quickly say "I have data at 9am on Thursday, I want data from 4pm Wednesday to 8am Thursday to create a model".

I figured the best way to do this efficiently was to have the code create a massive database for all the stocks and query it efficiently in SQL. But I'm stuck there due to a lack of tools.

Your problem sounds quite similar to a lot of FinTech interview questions for software engineers. They're solved problems but solutions aren't easy.

I have no doubt I could solve your problems. I honestly don't care to do so here though.

I imagine there's a software development/engineering team at your work or school you could ask for guidance though.

> I have no doubt I could solve your problems. I honestly don't care to do so here though.

That's fair. I did not think it'd be such a difficult problem when I first set out to do it myself. But every single turn leading to a dead end kinda bummed me out. I'm going to finally resort to the database method of storing all the data in one query-able file and work off of that.

You have 250GB of "raw" data stored in CSV format. The parsed version of this data in memory is likely to be a fraction of the on-disk size. A `long` or `double` only take up eight bytes in memory but 10-20 bytes on disk stored as ASCII in a CSV file. Even if your raw data was 250GB you could store it in memory mapped files. A fast SSD can easily hit a gigabyte per second sequential read speed, far faster than your typical network.

Segmenting your raw data and using memory mapped files will let you work with large data sets without needing huge amounts of RAM. From there it's a question of your single system's processing speed and IO capacity. This is only necessary if your processing needs random access to the entire dataset.

If your CSV data is more like a streaming data source, you're processing each record as it's read in, you can just stream it in through `stdin`. At 1GB/s you're looking at five minutes or so to process your 250GB of raw data. A SATA SSD might take twenty minutes to stream that raw data.

> A fast SSD can easily hit a gigabyte per second sequential read speed, far faster than your typical network.

It's important to note that often your disks aren't directly attached to your compute. That's frequently the case in (particularly cheap) cloud instances.

so, a hundred of the cheapest cloud instances cost you what? 1000 USD per month? You can easily get a deskside workstation for 10000 USD with a nice 2TB NVME-SSD, a 32-Core Threadripper and 128GB of memory... Utilization might be lower of course, but even if it's only utilized 8h/day this seems like a bargain for any solid business. For any startup aiming for exponential growth by burning through cash and having a 15month "half-life" this is not gonna work out though

That's the thing, you don't necessarily need a bunch of cloud instances to process data. If you must be doing everything in "the cloud" every service has dedicated instances with fast attached storage available. You can spin up one long enough to rip through your data instead of trying to distribute it over hundreds of workers.

It's also a domain where you can buy an off-the-shelf desktop for a few hundred dollars to do the work. That's the thrust of this whole thread, because scalable "cloud" systems exist and look cheap people obsess about throwing more instances at problems.

Modern commodity systems are ridiculously powerful and far more capable than people tend to assume. Even "the cloud" gets underestimated because people look at the low end cheap instances and assume they need to spin up hundreds of those when one beefy image for a short duration could do the same work.

> A fast SSD can easily hit a gigabyte per second sequential read speed, far faster than your typical network.

Nit: 1GB/s is ok, not even solid let alone fast.

A fast SSD pretty much saturates 4x3.0 links (which explains why they universally tend to cap out at 3.5GB/s). In fact there are now a few PCIe4 SSDs (e.g. Corsair's MP600) which close in on 5GB/s.

I got the 1GB/s from my laptop I was writing the comment on. The internal NVME drive tops out at 1.5GB/s. I consider that fast but as you point out there's drives that make mine look slow.

If there are floating point numbers in those csvs, I sped up a system like that 10x just by writing a custom (Java equivalent of) atof() that didn't do variable decimal separators and scientific notation. That's not even counting the improvements in I/O speed from the size reduction. Any system that works from CSV's is going to be slow. I don't know what sort of computations you're doing of course, but I did all the work on a laptop, couldn't be bothered to scale it out after the improvements I made in the first pass. How much of your code spends its time in I/O (including conversions) vs actually calculating?

Need to understand your domain better, but in many cases, the 250GB csv can be compressed down quite effectively using a columnar representation. And the columns can (potentially) be processed using simd/gpu based approaches to where a single server would outrun a cluster. Food for thought..

Two. The number is two. You always need a backup server :)

I used to think that. But there are a couple of notable exceptions at either end of the latency spectrum.

If your latency requirements are slack, then you can get away with one machine, because you can reboot or reprovision it and carry on processing without meet your requirements.

If your latency requirements are tight, you don't have time to fail over anyway, so you might as well run one machine and make sure you can deal with failing to meet your requirements.

This is why I ran Redis on a single server with no failover strategy. Our team spent maybe 15 hours workshopping a handful of other things (redis cluster, sentinels + replicas) before realizing we could spend an hour to have everything sit in a degraded state working around it while Redis itself was fixed. Redis only failed once ever and it took all of an hour to fix it, all during non-peak hours.

Three. So you can do maintenance on one while still having HA.

This guy SRE-s!

No, that's still two. You don't do maintenance when you're not awake.

*Sometimes three.

Go back to two when maintenance is done.

I was expecting a "two is one and one is none" reference. Three keeps the investors happy in my case.

But isn't the containerization trend leading us to a completely opposite direction ie. scale-out by default and do that not only because of performance but mainly because of how you want to manage your production environment?

Reply on news.ycombinator.com