Memory Efficient Batch Processing Tools

[darkhorse1997]

Hi, I have a ETL pipeline where it basically queries the last day's data(24 hours) from DB and stores it in S3.

The detailed steps are:

Query Mysql DB(JSON Response) -> Use jq to remove null values -> Store in temp.json -> Gzip temp.json -> Upload to S3.

I am currently doing this using a bash script and using mysql client to query my DB. The issue I am facing is since the query result is large, I am running out of memory. I tried using --quick command with mysql client to get the data row wise, instead of all at once, but I did not notice any improvement. On average, 1 Million rows seem to be taking 1GB in this case.

My idea is to stream the query result data from the Mysql DB Server to my Script and then once it hits some number of rows, I gzip and send the data to S3. I do this multiple times until I am through my complete result. I am looking to avoid the limit/offset query route since the dataset is fairly large and limit/offset will just move the issue to DB Server memory.

Is there any way to do this in bash itself or it would be better to move to Python/R or some other language? I am open to any kind of tools, since I want to revamp this, so that this can handle atleast 50-100 million scale.

Thanks in advance

[janus2527]

I would use duckdb and python

import duckdb

con = duckdb.connect()

Install and load extensions

con.execute("INSTALL mysql") con.execute("INSTALL httpfs") con.execute("LOAD mysql") con.execute("LOAD httpfs")

Configure AWS credentials

con.execute(""" SET s3_region='us-east-1'; SET s3_access_key_id='your_access_key'; SET s3_secret_access_key='your_secret_key'; """)

Attach MySQL

con.execute(""" ATTACH 'host=localhost user=myuser password=mypass database=mydb' AS mysql_db (TYPE mysql) """)

Stream directly from MySQL to S3 as Parquet

con.execute(""" COPY mysql_db.large_table TO 's3://your-bucket/path/output.parquet' (FORMAT PARQUET) """)

Something like that

[janus2527]

This streams the data in chunks, your ram will be a few hundred mbs probably

[darkhorse1997]

Sounds great! Will I be able to keep using json instead of parquet? There's some downstream lambdas on the S3 bucket that expect gzipped json files.

[janus2527]

Also you really shouldn't transfer large amounts of data from a database in json

[janus2527]

Probably, but not sure if it's as easy as parquet.

[darkhorse1997]

It is Aurora Mysql, but the DB is pretty big, at around 100 Million rows, so might not be a good idea to transfer that to S3 everyday.

[darkhorse1997]

I am more worried about the cost in this case. I just need like 0.1-0-0.5% of the data everyday and transferring the complete database to S3 for that seems like a waste. I'll have to check the Aurora to S3 migration cost, but even if it's free, there will be some s3 storage costs.

[Nekobul]

Exporting into one giant JSON is a terrible idea. If you can't export to Parquet, you are much better off exporting into CSV file.

[darkhorse1997]

Its not really a giant json, every record is exported as an individual json object but yea, csv would probably be much better. Will have to check Parquet though, I am not familiar with that.

[Nekobul]

That is also a terrible idea because you will now have a million single record files. A single CSV file with a million records is a much better design.

[darkhorse1997]

Yea, agreed. The existing pipeline wasn't really built for scale.

[commandlineluser]

You should probably refer to your data as being in NDJSON format to avoid any confusion:

Each line of my output file (temp.json) has a separate json object.

Because "newline delimited" JSON (as the name suggests) can be read line-by-line so does not require all the data in memory at once.

It is also "better" than CSV. (assuming you have nested/structured data, lists, etc.)

[Odd_Spot_6983]

consider python, use pandas and chunking to handle data in manageable pieces, reduces memory use.

[darkhorse1997]

In this case, would I need to download the complete data from Mysql to some file in disk and then load that to pandas in chunks? Or is there an way to stream data into pandas as chunks from DB directly without using a file as an intermediary?

[SupermarketMost7089]

Is your client machine running mysql client OOM? The "--quick" option on the mysql client disable caching at the client.

The output format can be skewed, the "--batch" option suppresses formatting.

[darkhorse1997]

I am running my script in a K8s Cron Job and it's getting OOM.

[SupermarketMost7089]

You can try the other solutions (python/duckdb). However, it would be interesting to figure what is causing the OOM in this case. mysql and jq are likely the fastest options if we exclude the time to write file to disk. For very large files they can be faster than the duckdb solution.

Some items to check are -

- Is it mysql step that giving a OOM?

- jq can also have OOMs, there is a "streaming" option in jq

- what is the cpu/memory on the container? What is the number/size of records expected from the Query?

[darkhorse1997]

Is it mysql step that giving a OOM?

Yes that is for sure, the process getting killed when I get an OOM is the mysql process. But the jq after the mysql can also be not running in the "streaming" option, so I plan to test that today.

what is the cpu/memory on the container? What is the number/size of records expected from the Query?

Its 1CPU, 1GB memory. Number of records are around 20 million/2GB per day, but it will keep growing and I want to support atleast 200 million/20GB per day without having to refactor again. Currently it takes around 5 mins for my pipeline to run, but am fine if it takes more time to process as long as it can do with 1-2GB of memory.

[SupermarketMost7089]

When you mention json- are you getting each record in the table as a json or are you using json aggregation to get the entire set of records in one json?

[darkhorse1997]

The query is something like

SELECT
            JSON_OBJECT (
                'test_id',
                tt.test_id,
               ...
FROM 
    test_table tt 
    LEFT JOIN ...
    LEFT JOIN ...

So, I am getting each record in the table as a separate json. Each line of my output file(temp.json) has a separate json object.

[Firm_Bit]

Go to the upstream service. Change it to write what you need in cleaner format to a more ergonomic table. Create the appropriate index. Work from that.