[Python] Optimize INSERT from staging table into spatial table

Background: I use airflow to orchestrate an ETL process which moves data from snowflake to s3 and from s3 to Postgres. The entire process moves around 80 million rows of data from snowflake to Postgres. In Postgres the data is loaded into a staging table before being transformed and inserted into its prod table.

Problem: The insert from staging into the prod table creates a geopoint using ST_MakePoint(longitude, latitude). This process plus the indexing that happens after the insertion take the majority of the duration of this ETL.

Sample code:

-- Insert data from staging table to production table
INSERT INTO public.target_table (geo_lookup, proximity_data, updated)
SELECT
ST_MakePoint(longitude, latitude) as geo_lookup,
proximity_data,
CURRENT_TIMESTAMP as updated
FROM public.source_table;

-- Create a GiST index on the geo_lookup column
CREATE INDEX IF NOT EXISTS target_table_geo_idx
ON public.target_table
USING GIST (geo_lookup);


Environment details

PostgreSQL: 16.1 on aarch64-unknown-linux-gnu
PostGIS: 3.4
Airflow: 2.9.2 (managed on AWS MWAA)
Airflow environment class: mw1.small
Scheduler count: 2
Worker count: min 1, max 10
Web server count: min 2, max 2


So far I have tried the following:

1. Batch insert rows with autocommit instead of the single transaction I currently use (still sequentially loops over each batch to insert though so not concurrent). The result was actually worse than what I currently do, increasing the duration of the DAG significantly.
2. Increasing hardware spec on aws rds serverless instance to see if overall DAG duration would reduce. This had almost no impact of the duration of the DAG. The tasks in airflow that handle the postgres insert and index took roughly the same amount of time.

Things I haven't tried yet are setting up some form of concurrency either through python async or python multiprocessing/multithreading. Although I am not entirely sure how that would work with airflow since I would need to make parallel tasks for the insert operation? Or I would just create a single python operator that at run-time would handle doing the async or multiprocessing. If anyone has any examples of something like this or any direction I am all ears. Or I could try to approach it from the snowflake side and figure out if there is a way I can track what data is changed there and only insert that smaller subset of data into postgres. Again would just be curious if anyone can point in the right direction or have any examples on how any of these things can be tackled. Thanks in advance.

def _insert_and_index_geo_data(self):
insert_sql = """
INSERT INTO public.target_table (geo_lookup, proximity_data, updated)
SELECT
ST_MakePoint(longitude, latitude) AS geo_lookup,
proximity_data,
CURRENT_TIMESTAMP AS updated
FROM public.source_table;
"""
self.pg_conn.execute_sql(query=insert_sql, id='insert_geo_data')

index_sql = """
CREATE INDEX IF NOT EXISTS target_table_geo_idx
ON public.target_table
USING GIST (geo_lookup);
"""
self.pg_conn.execute_sql(query=index_sql, id='create_geo_index')

Continue reading...