Distributed processing of a PostgreSQL table
<p>I've got a PostgreSQL table with several millions of rows that need to be processed with the same algorithm.
I am using Python and SQLAlchemy.Core for this task.</p>

<p>This algorithm accepts one or several rows as input and returns the same amount of rows with some updated values.</p>

<pre><code>id1, id2, NULL, NULL, NULL -&gt; id1, id2, value1, value2, value3
id1, id3, NULL, NULL, NULL -&gt; id1, id3, value4, value5, value6
id2, id3, NULL, NULL, NULL -&gt; id2, id3, value7, value8, value9
...
id_n, id_m, NULL, NULL, NULL -&gt; id_n, id_m, value_xxx, value_yyy, value_zzz
</code></pre>

<p>I am using a PC cluster to perform this task. This cluster runs <code>dask.distributed</code> scheduler and workers.</p>

<p>I think, that this task can be effectively implemented with the <code>map</code> function. My idea is that each worker queries data base, selects for processing some rows with NULL values, then updates them with results.</p>

<p>My question is: how to write the SQL query, that would allow to distribute pieces of the table among workers?</p>

<p>I've tried to define subsets of row for each worker with <code>offset</code> and <code>limit</code> in SQL queries, that each worker emits:</p>

<p><em>SQL:</em></p>

<pre><code>select * from table where value1 is NULL offset N limit 100;
...
update table where id1 = ... and id2 = ...
       set value1 = value...; 
</code></pre>

<p><em>Python:</em> </p>

<pre><code>from sqlalchemy import create_engine, bindparam, select, func
from distributed import Executor, progress

def process(offset, limit):
    engine = create_engine(...)

    # get next piece of work
    query = select(...).where(...).limit(limit).offset(offset)

    rows = engine.execute([select]).fetchall()
    # process rows

    # submit values to table
    update_stmt = table.update().where(...).where(...).values(...) 
    up_values = ...
    engine.execute(update_stmt, up_values) 

if __name__ == '__main__':
    e = Executor('{address}:{port}'.format(address=config('SERVER_ADDR'),
                                       port=config('SERVER_PORT')))
    n_rows = count_rows_to_process()
    chunk_size = 100
    progress(e.map(process, range(0, n_rows, chunk_size)))
</code></pre>

<p>However, this didn't work. </p>

<p>The <code>range</code> function has returned list of offsets before calculations have started, and the <code>map</code> function has distributed them among workers before starting <code>process</code> function. </p>

<p>Then some workers have successfully finished processing their chunks of work, submitted their results to the table, and updated values. </p>

<p>Then new iteration begins, new <code>SELECT ...WHERE value1 is NULL LIMIT 100 OFFSET ...</code>  query is sent to the data base, but the offset is now invalid, because it was calculated before the previous workers have updated the table. Amount of NULL values is now reduced, and a worker can receive empty set from the database.</p>

<p>I cannot use one <code>SELECT</code> query before starting calculations, because it will return huge table that doesn't fit in RAM.</p>

<p>SQLAlchemy manual also says that for distributed processing the engine instance should be created locally for each python process. Therefore, I cannot query the database once and send returned cursor to the <code>process</code> function.</p>

<p>Therefore, the solution is correct construction of SQL queries.</p>