[Python] How to perform parallel data migration using Python

I want to perform data migration between tables in parallel. For example, I have data for cities A, B, and C. Let's say the data migration takes about 30 minutes. Migration of cities A and B is done at 1 am. Then city C at 1.10 am.

In the code that I wrote, migration cannot run in parallel. Migration of city C will be done after cities A and B are finished. How can I do that? The following is the generated data that I created:

import pandas as pd import random

# Defining column names
columns = ['ID', 'Name', 'City', 'City_ID', 'Revenue']

# List of cities and corresponding city IDs
cities = {
'New York': 1,
'Los Angeles': 2,
'Chicago': 3,
'Houston': 4,
'Phoenix': 5 }

# Generating 1000000 rows of dummy data
data = {
'ID': range(1, 1000001),
'Name': [f"Customer_{i}" for i in range(1, 1000001)],
'City': [random.choice(list(cities.keys())) for _ in range(1000000)],
'Revenue': [round(random.uniform(20.0, 500.0), 2) for _ in range(1000000)] }

# Creating City_ID based on the City column
data['City_ID'] = [cities[city] for city in data['City']]

# Creating DataFrame
df = pd.DataFrame(data)

# Displaying the first few rows of the DataFrame
print(df.head())

# Optionally, you can save the data to a CSV file
df.to_csv('D:/INSAN/Document Project/dummy_data.csv', index=False)


city_flagging table

1 New York 0 0 0
2 Los Angeles 0 0 0
4 Houston 0 0 0
5 Phoenix 0 0 0
3 Chicago 1 0 0

Here is the code for data migration:

import pyodbc
import datetime
import time
import concurrent.futures

# Save migrated cities
migrated_city_ids = set()

# Function to get is_available = 1 status from database
def check_flagging_status():
conn = pyodbc.connect('Driver={PostgreSQL ODBC Driver(UNICODE)};server=localhost;port=5433;database=postgres;uid=postgres;pwd=postgres')
cursor = conn.cursor()

query = "SELECT * FROM city_flagging WHERE is_available = 1 AND is_running = 0"
cursor.execute(query)

available_cabangs = cursor.fetchall()

cursor.close()
conn.close()

return available_cabangs

# Function to update is_running and is_success status in city_flagging table
def update_flagging_status(city_id, is_running=None, is_success=None):
conn = pyodbc.connect('Driver={PostgreSQL ODBC Driver(UNICODE)};server=localhost;port=5433;database=postgres;uid=postgres;pwd=postgres')
cursor = conn.cursor()

try:
if is_running is not None:
query = "UPDATE city_flagging SET is_running = ? WHERE city_id = ?"
cursor.execute(query, (is_running, city_id))

if is_success is not None:
query = "UPDATE city_flagging SET is_success = ? WHERE city_id = ?"
cursor.execute(query, (is_success, city_id))

conn.commit()
except Exception as e:
print(f"Error while updating flagging status for city {city_id}: {e}")
finally:
cursor.close()
conn.close()

# Single city data migration function
def migrate_data(city_id, city_name):
if city_id in migrated_city_ids:
print(f"City {city_id} already migrated before, skip migration.")
return

# Update is_running to 1 (migration process is running)
update_flagging_status(city_id, is_running=1)

date = datetime.datetime.now().date()
start_datetime = datetime.datetime.now()

conn_old = pyodbc.connect('Driver={PostgreSQL ODBC Driver(UNICODE)};server=localhost;port=5433;database=postgres;uid=postgres;pwd=postgres')
cursor_old = conn_old.cursor()

conn_new = pyodbc.connect('Driver={PostgreSQL ODBC Driver(UNICODE)};server=localhost;port=5433;database=postgres;uid=postgres;pwd=postgres')
cursor_new = conn_new.cursor()

try:
query = f"SELECT * FROM dummy_data WHERE city_id = '{city_id}'"
cursor_old.execute(query)
transaksi_data = cursor_old.fetchall()

print(f"Data migration for city {city_id} is in progress.")
insert_query = "INSERT INTO dummy_data_1 (id, name, city, revenue, city_id) VALUES (?, ?, ?, ?, ?)"
cursor_new.executemany(insert_query, transaksi_data)

conn_new.commit()

end_datetime = datetime.datetime.now()

log_query = """
INSERT INTO migration_log (date, city_id, city_name, start_datetime, end_datetime)
VALUES (?, ?, ?, ?, ?)
"""
cursor_new.execute(log_query, (date, city_id, city_name, start_datetime, end_datetime))
conn_new.commit()

# Update is_success menjadi 1 (migrasi berhasil)
update_flagging_status(city_id, is_success=1)

migrated_city_ids.add(city_id)
print(f"Data migration for city {city_id} is complete.")

except Exception as e:
print(f"Error while migrating data for city {city_id}: {e}")

finally:
cursor_old.close()
conn_old.close()
cursor_new.close()
conn_new.close()

# Function to start migration
def start_migration():
city_list = check_flagging_status()

if not city_list:
print("There are no cities available for migration.")
return False # Tidak ada city yang tersedia

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
for city in city_list:
city_id = city[0]
city_name = city[1]
executor.submit(migrate_data, city_id, city_name)

return True # There is a migration underway

# Function for periodic checks
def continuous_check(interval_seconds=2, max_no_migration_attempts=3):
no_migration_attempts = 0

while no_migration_attempts < max_no_migration_attempts:
print("Checking flagging city status...")
migration_started = start_migration() # Check and migrate if there is a new city available

if migration_started:
no_migration_attempts = 0 # Reset trial if there is a migration running
else:
no_migration_attempts += 1 # TAdd trial if no city available

if no_migration_attempts >= max_no_migration_attempts:
print("No cities available for migration in multiple attempts. Stopping the process.")
break

time.sleep(interval_seconds) # Wait for the specified interval

# Run periodic checks in parallel
continuous_check(1, 3)

Continue reading...