• About
• Quick Setup
• How to use
• Configuration
• Approach
• Limitations
• Benchmarks
• Examples

Import records from a csv input to a existing postgres database without using the instruction COPY.

Execute the command and wait until the import process is complete:

python3 -m py_import -i path/to/your/file.csv

Clone this repo.

git clone https://github.com/sGaps/pg_import.git

Open the project folder

cd pg_import

If you have a testing database already, you may be interested in configuring the credentials files .postgres.* that are inside the credentials/dev folder. For example, if you want to edit the password used to connect to the database, then you should edit the file credentials/dev/.postgres.psw.

echo 'my-password' > credentials/dev/.postgres.psw

If you don't have any existing database, you can create a new one by executing

docker compose up -d

And you won't need to create the tables manually, pg_import will create the tables it needs to work.

After completing Quick setup, you can import some records using the module notation:

python3 -m py_import -i <path/to/file.csv>

or using the script utility:

python3 run_import.py -i <path/to/file.csv>

Note that you can cancel the process by pressing Ctrl+C in the terminal or sending a SIGTERM to the process. Before exiting, the program will delete all records it has inserted. If you press Ctrl+C before the rollback process ends, you will be able to preserve the changes made in the db. (but sometimes, the rollback will be so fast that you won't be able to cancel it).

Also, you may be interested in seeing some addditional examples.

This project loads its configuration from a directory named credentials/, which should contain a subfolder named dev, which can have several files that define the values used to create PostgreSQL Connections.

./
    credentials/
        dev/
            .postgres.db
            .postgres.user
            ...

For now, it seems that it's a little weird to have a folder with a subfolder to contain connection arguments. Well, it actually is!, but there's a reason for that.

pg_import accepts a parameter named staging, which instructs the program to load a bunch of parameters from another subfolder so that you can keep all those configurations in your workflow.

By default, staging has the value dev, so it can load the files inside credentials/dev. But you can specify the staging by setting the environment variable $PG_IMPORT_STAGING or also by adding a file named .staging in your current folder. (note that the env variable has precedence over the .staging file).

env PG_IMPORT_STAGING=qa python3 -m pg_import -i ...
# > will load the files inside: `credentials/qa`

Also, you can specify the directory that holds the staging folders by setting the environment variable PG_IMPORT_CONFIG_PATH, which has the value credentials as default.

Currently, the files that are loaded from the path {credentials}/{stagging} are:

• .postgres.db
• .postgres.psw
• .postgres.user
• .postgres.port
• .postgres.host

The goal was to insert more than 17M records without using the COPY directive that PostgreSQL implements, and using a program written in python to achieve it.

So, the first thing I needed was a way to create a database quickly. In this case, I used docker compose to configure a PostgreSQL service quickly, which can be run with the command:

docker compose up -d

Then, I added some docker secrets to manage the credentials and organize it into a basic configuration structure like the one seen in the configuration section.

Already having a database, the next step was to create a simple connection to the database using python so I considered to use the package psycopg because it's widely used for projects that involve Postgres databases. Also, I used psycopg2 before, so I thought it was good to experiment with a newer version of that package.

After that, I implemented a simple connection script to perform some queries and check whether the connection works fine. Later, I saw that I needed to create an initial schema, so I chose the package sqlalchemy to help me define the tables/models the program requires by using the declarative interface and ORM that this component provides.

Later, I designed the schema with two models: SaleOrder (indexed) which holds the data that we are trying to import and ImportRegistry, that keeps track of the records that were imported in SaleOrder. But why not to just add a simple Boolean to mark the records that were imported?

Well, It could have worked for this simple example

But if we tried to move this solution to a general case, we could have some problems with the existing database tables. To implement the Boolean solution, we need to add this field to every table that will implement an import scheme. This implies that we have to choose a name that doesn't cause any conflict with the existing columns of all tables, which is hard.

In contrast, having a separate table that keeps track every import is more suitable for a growing application. If we want to track imports on other tables, we could just add a new attribute in this model that holds the name of a table where an import process has been performed. And as we know that we only import massive amounts of data, I decided to have the records start_id and end_id to represent import ranges.

Having the schema and both packages included in the project, I made some simple scritps that implemented a brute force solution, which consisted on uploading all records at once. The purpose of this approach was exploring how many resources the program needs to process the whole sample.

After almost running out of memory, I canceled the process and saw that It could upload 6.9M of records by using 7GiB of RAM. Knowning the limitations the brute force approach had, I realized that a huge improvement was needed in order to come up with a real solution, so I started analyzig what went wrong.

The first thing I detected was that the original scripts were creating ORM records in memory and saving them in the session pool. Each time we created a new record, the memory usage of the session was increasing as well until reaching the point where the program couldn't allocate more memory. With that in mind, we could just divide the sample into chunks and send them to the database.

To support this solution proposal, I designed an incremental import processing, where the program loads chunks of 100k records and commits the changes after that. But as I was still using the ORM, it took around 18 minutes and 4GiB of RAM to complete the process.

Having a minimal working program, I decided to implement a simple but useful CLI interface to make easier performing the tests. During this step, I thought that pg_import needed to be composable, so it should be able to read from stdin, and as it was accepting csv files, it also needed a way to specify the column delimiters used in the input. The right tool to create a simple CLI was the module argparse, which is much simpler than the getopts alternative.

Coming back to the harder problem, I performed an import test with the command psql and its built-in instruction \copy to see how fast it can load the sample of 17M records. The test took 1 minute and few seconds. It seems that COPY uses a large set of optimizations to ensure that the data will be available as soon as possible.

Knowing that the challenge is to avoid using this bulk-insert operation, I needed to follow some additional strategies to provide an efficient and scalable solution.

I decided to drop the fragment of code that relied on the SQLAlchemy's ORM for the insert operations and decided to use a db cursor directly. This is possible because we are only interested in submit data in a PostgreSQL instance, so we don't worry about the compatibility layer among database engines that the ORM provides.

After rewriting the ORM queries as raw SQL queries, the execution time improved dramatically. It took around 12 minutes to complete the task, although, the program was not fast enough.

I knew that there was room to keep improving the code, so I started a research on the psycopg documentation, looking for parameters, configurations and utilites that could improve the performance even more. After a while, I came accross a fragment of the documentation which said that since the third major version of the package, every conection created by the database engine used Server-sided cursors by default. This kind of cursor offers more reliability and let library users to write pythonic code with results that come from a cursor.execute call.

The Server-sided cursors also split the queries that have more than a single record in it, which causes an overhead and make harder to define efficient queries with raw sql. So I decided to switch to Client-sided cursors. These, are the same that psycopg2 uses, and they provide an interface to write queries without limiting the amount of records sent and returned as the other cursor type does. The previous benefits come at the cost of compromsing the security abstraction layer that offered by the new cursor type. (An additional comment about this matter can be found in the commit 897960a8e34ad3be5fb7c3f9faeca43e7f3537f3).

After inserting the Client-sided cursors on the bulk insertions and deletions, the process took much less time to complete the task: 8 minutes and 53 seconds. Unfortunately, deleting the records from the previous import process took too much time in the next program execution (after 10 whole minutes, I canceled the delete operation), so I noticed that I needed to change the query in order to optimize the execution time.

Later, I tested the program and it took 8 minutes and 45s to finish, including the deletions that must be performed before inserting the new records.

Now, having the program that solves the problem, I started to tune it by changing the value of a constant named CHUNK_SIZE until I found the optimal size that the chunk must have in order to consume the minimal amount of resources.

Finally, I started to clean up the code, add documentation, test the code to check that it's working properly, and finally, adding some features to improve the CLI interface.

SIDE NOTE: We could have used temporal unlogged tables among the one used to store the actual data, but this kind of table would be truncated if the database fails in the middle of the process (for example, if there's a power outage).

This project was made only for educative purposes, so the program assumes that the csv always have the same csv format like the one described in the following table:

which can be translated to a plain csv file similar to:

PointOfSale;Product;Date;Stock
POS-TEST;PRODUCT-TEST;2019-08-17;10
...

Environment used for the test:

• Hardware
  • CPU: i5-10400
  • RAM: 2x8 GiB @ 2400 MT/s
  • SWAP: 16GiB (HDD)
• Software:
  • OS: Manjaro 23.0
  • Kernel: 5.15.131-1

The parameter that is being tested is CHUNK_SIZE. This indicates the size of record frames that we send to the database to perform an incremental create/update/delete. After experimenting with several settings for this parameter, we obtained the following results:

Show help

python3 -m pg_import

Read csv from stdin

python3 -m pg_import

Read csv without header fom stdin

python3 -m pg_import --no-header

Reading from file

python3 -m pg_import -i values.csv

Alternatively

python3 -m pg_import --input values.csv

Specifiying separator or delimiter (it has the value ';' by default).

python3 -m pg_import -d , values_with_comma.csv

Or also works with

python3 -m pg_import --delimiter=, values_with_comma.csv

Combining pg_import with other shell utilities. In this example, we remove the header of the file, with tail, take the first 100 records, sort them and finally, importing them into the database.

tail -n +2 values.csv | head -n 100 | sort | python3 -m pg_import --no-header 

Using the run_import.py script instead of the module pg_import

python3 run_import.py -i values.csv

Using connection settings from a specific staging:

env PG_IMPORT_STAGING=QA python3 -m pg_import -i values.csv

Using custom path and staging for the connection settings:

env PG_IMPORT_STAGING=QA PG_IMPORT_CONFIG_PATH=../different/location/credentials python3 -m pg_import -i values.csv