GPU Data Engineering, CSV to Parquet & ORC using Dask & RAPIDS.ai
Converting large CSV files to Parquet and ORC formats using Dask and RAPIDS.ai on consumer GPU hardware, with performance benchmarks and optimization strategies.
Up until recently the majority of my data engineering work was around designing and deploying Data pipelines in either AWS or GCP, there hasn’t been much need for me to delve into the GPU side of things for a while. As I recently find myself with a bit of time on my hands I thought I would revisit my previous experiments that I detailed in this previous article:
GPU Datascience: Converting CSV to Parquet / ORC faster & Cheaper than using a cluster!
The article above goes into the basics of why these particular experiments were chosen, a little about the dataset and the setup I am using.
I chose to use the same setup and dataset to provide some form of comparison, but hopefully I can use better methods to convert the data, this will hopefully mean that any increases seen (or decreases) are to do more with the underlying library or filetype rather than the hardware or the data.
The Hardware
My NAS is getting on a bit now, its not ancient but due to the huge jump in GPU sizes and power requirements, it is kind of limited as to what I can fit in it, I do have a GTX 1080 which will give a bit more power that fits but actually it makes sense to run it on a lower memory model (will discuss later).
Preparing the data
OK, at some point in the past I decided to convert my test data into AVRO and then promptly delete the original CSV data, so first things first — I need to convert it back.
!du -hs /ssd/Vegas/datafiles
85G /ssd/Vegas/datafiles
%%time
path = "/ssd/Vegas/datafiles/lv*"
for filename in glob(path):
p,f = os.path.split(filename)
df = cudf.read_avro(filename)
ddf = dask_cudf.from_cudf(df, npartitions=3000)
ddf.to_csv(f"/ssd/Vegas/datafiles/csv/{f}.csv", single_file = True)As before this needs to be heavily partitioned to fit into the GPU memory, however the recent introduction of “single_file” in the dask to_csv is a nice touch as it means I don’t have to deal with 3000 CSV files for each AVRO file.
This was incredibly slow — gave up after it had only done 48 out of 630 files — I need to parallelise it more — there should be some new updates to help since I last used cudf & dask
After reading the docs I realised I can probably leverage LocalCUDACluster to be able to have dask chunk the data as required or spill to system memory, so I decided to give this a go and use the dask_delayed functionality to allow me to pre-build the cudf dataframes and then process them quickly with spill to memory preventing the OOM errors.
First I will set up a LocalCUDACluster and set some memory limits:
%%time
cluster = LocalCUDACluster(
CUDA_VISIBLE_DEVICES="0",
rmm_pool_size=parse_bytes("5.5GB"), # This GPU has 6GB of memory
device_memory_limit=parse_bytes("4.5GB"), # 75% of total
)
client = Client(cluster)
clientpythSo first we are simply going to iterate through the files on disk and build cudf dataframes for later.
%%time
path = "/ssd/Vegas/datafiles/lv*"
def load_files(filename):
df = cudf.read_avro(filename)
return dfThen we are going to setup the Dask pipeline to build some delayed dataframes from the file list
%%time
delayed_dfs = [delayed(load_files)(file) for file in glob(path)]
ddf = dc.from_delayed(delayed_dfs)
CPU times: user 666 ms, sys: 245 ms, total: 912 ms
Wall time: 3.13 sAnd then finally run the delayed dask_cudf to create the CSV files:
%%time
ddf.to_csv(f"/ssd/Vegas/datafiles/csv/*.csv")
CPU times: user 40.1 s, sys: 12.7 s, total: 52.8 s
Wall time: 22min 2sLooking at the dataset, there is something wrong — its only 188GB which means its entirely possible that I only converted some of the CSV files to AVRO and then deleted the rest as previous data was around 400GB uncompressed.🤦♂️
Data Augmentation
This is usually a fairly involved process that involves training models and GaN’s to produce good test data when yours is lacking, but as this is just a simple volume test I can get away with just mangling some of the other data to create some entropy:
%%time
delayed_csv_dfs = [delayed(load_csv_files)(file) for file in glob(path)]
cddf = dc.from_delayed(delayed_csv_dfs)
copy_cddf = cddf.copy()
cddf.id = str(uuid.uuid1())
cddf.type = random.choice(['aaid','idfa'])
...
CPU times: user 65.1 ms, sys: 14.9 ms, total: 80 ms
Wall time: 378 msand then writing these copies back to storage to make up the remainder of the data:
%%time
cddf.to_csv(f"/ssd/Vegas/datafiles/csv/aug*.csv")
CPU times: user 40.1 s, sys: 12.7 s, total: 52.8 s
Wall time: 22min 2sThis gives us a nice 394.8G which is the same size as the previous dataset (and roughly the same time to convert this(40 mins) as previous benchmarks if added together 🤔)
OK! to work!
So first things first, why am I using an old and outdated GTX 1060 card with 6GB ram when for a few bucks I could run this on the latest and greatest?
GPU’s are not designed to have huge amounts of memory, even the largest GPU’s do not have enough memory to work with the largest datasets without some wrangling and the best way to demonstrate this is to have very little available memory on what is basically the lowest supported card by RAPIDS.ai, think of it as “the most painful demonstration”.
My little 6GB of memory in my consumer GTX 1060 is the ideal card to make sure that I am not just getting good results because my data is “convenient”, I want to create some examples that will work on examples 10x the size on better hardware. If I can do it on this, theoretically, I can do it on anything better too!
I am also a glutton for punishment and I know that this is going to involve some frustrating out of memory errors when I start to do analysis on things, I find that when you end up hitting problem after problem you learn a lot and very quickly, plus, I seem to have made a career of hitting sharp edges on bleeding edge technology so why not continue to do it here?
I am using the same base setup for reading the CSV files as I did for the Augmentation above and I will just add the results of the write to Parquet, ORC.
CSV to ORC
This was the same test I did last time so it seemed like a good place to start — 11 minutes — exactly the same time as the previous test with exactly the same setup
However what was different this time was not needing to hugely partition the data to fit into GPU memory meaning that I am not dealing with 100k+ files — it just “worked” which is a much cleaner workflow than before.
%%time
path = "/ssd/Vegas/datafiles/csv/*.csv.gz"
#drop unnecessary columns
cols_to_use = ['id', 'ts', 'id_type', 'lon', 'lat', 'ha']
ddf = dc.read_csv(path, usecols=cols_to_use, blocksize="500MB").to_orc(f"/ssd/Vegas/datafiles/orc/vegas")CPU times: user 1min 13s, sys: 20.1 s, total: 1min 33s
Wall time: 31min 46sUnzipped CSV Files to ORC
GZIP is not splitable so I assume there is some overhead when operating on csv.gz files that I also wanted to test. Turns out it is a pretty huge overhead, so if space it not important to you, operate on the CSV files for a 300% speed up!
11 minutes for 400GB+ is nice!
path = "/ssd/Vegas/datafiles/csv/*.csv"
#drop unnecessary columns
cols_to_use = ['id', 'ts', 'id_type', 'lon', 'lat', 'ha']
ddf = dc.read_csv(path, usecols=cols_to_use, blocksize="500MB") \
.to_orc(f"/ssd/Vegas/datafiles/orc/vegas")CPU times: user 25.7 s, sys: 7.38 s, total: 33.1 s
Wall time: 11min 37sCSV to Parquet
This came out about the same time as in 2019 converting to ORC on SATA (I didn’t try and convert to Parquet back then) — 40 minutes — but this was on NVME drives so this is a little disappointing. Certainly slower than the ORC conversion (Its possible that Parquet pre-calculates a lot more than ORC when creating it — I guess we will see if it pays off during the analysis!) — and again — no partitioning needed to get the result.
%%time
ddf.to_parquet(f"/ssd/Vegas/datafiles/parquet/vegas", write_index=False)
CPU times: user 1min 16s, sys: 23.6 s, total: 1min 39s
Wall time: 40min 6sI mean, 40 minutes on consumer hardware for maybe a few cents of electricity is still not bad, its certainly doable in terms of the size of data we are working with
Unzipped CSV Files to Parquet
%%time
cols_to_use = ['id', 'ts', 'id_type', 'lon', 'lat', 'ha']
ddf = dc.read_csv(path, usecols=cols_to_use, blocksize="500MB") \
.to_parquet(f"/ssd/Vegas/datafiles/parquet/vegas")
CPU times: user 38.8 s, sys: 11.7 s, total: 50.4 s
Wall time: 19min 58s19 minutes for 400GB+ — not bad!
This came out at half the the time, not as impressive as the ORC results but still worthy!
Conclusion
So what does this all mean?
I guess I was expecting some kind of magic to have happened with the libraries over the past 3 years (RAPIDS.ai was in a very early release back then), that these updates would somehow makes my grandpa GPU work faster or more efficiently but I guess previously tests in 2019 were basically using 100% of its capabilities already (or I am still being horribly inefficient somehow which wouldn’t be surprising).
It is after all a very simple use case that pretty much all data engineers have to do at some point — of course the Rapids team optimised for this.
The results on unzipped CSV files were impressive, obviously if space considerations are there this might not be feasible, I would like to repeat the experiment with SNAPPY or XZ as the compression method (maybe a later article) as that can be split, perhaps there is a decent trade-off between storage and speed that can be found.
I did hit some painful realisations as to the method I have described above during analysis which I will go into in the next article, I will link to that here when I have finished (assuming I have any hair left afterwards), but I think for now I think I can put the “CSV to <format>” methods to bed as being about as efficient as they can be on this hardware.
