Well, thanks to the all help -> I was able to painlessly create a postgreSQL database quite easily using the below code.
However, it hasn't really helped because it still takes days and days to perform the group by.
I'm not really sure of a faster way of doing this; would creating 20 indexes first help? It seems the fastest way to collapse is to keep the data in its disaggregated form.
def call_robocopy(from_folder='',
to_folder='',
my_log='H:/robocopy_log.txt'):
"""
Copy files to working directory
robocopy <Source> <Destination> [<File>[ ...]] [<Options>]
"""
if os.path.isdir(from_folder) & \
os.path.isdir(to_folder):
call(["robocopy", from_folder, to_folder, "/LOG:%s" % my_log])
else:
print("Paths not entered correctly")
def pandas_temp_table(path_2_csv='',
tmp_table='',):
"""
Upload data to a temporary table first using PANDAs to identify optimal data-types for columns
PANDAS is not speed-efficient as it uses INSERT commands rather than COPY e.g. it took COPY 16mins average
to get a 15GB CSV into the database (door-to-door) whereas pandas.to_sql took 50mins
"""
# Pandas can use sqlalchemy engine
engine = create_engine('postgresql://%s:%s@localhost:5432/%s' %(myusername, mypassword, mydatabase))
if engine:
print('Connected: %s' % engine)
else:
print('Connection lost')
sys.exit(1)
tmp_table += '_temp'
counter = 0
start_time = time.time()
for i in os.listdir(path_2_csv):
# Cycle through all CSVs and upload a small chunk to make sure everything is OK
if counter < 1:
if i.endswith(".csv") & i.startswith("100_pct"):
print("Reading CSV: %s into PANDAs data-frame" % i)
# First 1,000,000 rows
#df = pd.read_csv(os.path.join(path_2_csv, i), nrows=1000000, header=None, sep='~') #sep=None; automatically find by sniffing
# Upload whole file
df = pd.read_csv(os.path.join(path_2_csv, i), header=None, sep='~') #sep=None; automatically find by sniffing
df.columns = [
.. around 30 columns ..
]
print("CSV read-in successfully")
print(df.shape)
print("Uploading %s to SQL Table: %s" % (i, tmp_table))
df.to_sql(tmp_table, engine, if_exists='append', index=False)
counter += 1
current_speed = ((time.time()-start_time)/60)/counter
print("Average speed is %.2f minutes per database" % current_speed)
print("Successfully uploaded: %d" % counter)
end_time = time.time()
print("Total duration of INSERT: %.2f minutes" % (end_time - start_time)/60)
def create_postgresql_table(my_table=''):
"""
Create table copying the structure of the temp table created using pandas
Timer to benchmark
"""
# Connect
con = psycopg2.connect(database=mydatabase, user=myusername, password=mypassword)
cur = con.cursor()
if con:
print('Connected: %s' % con)
else:
print('Connection lost')
sys.exit(1)
try:
# Check if table exists already
cur.execute("""
SELECT relname FROM pg_class WHERE relname = '{0}';
""".format(my_table))
table_test = cur.fetchone()[0]
except Exception as e:
print('Table %s does not exist' % my_table)
table_test = None
if table_test:
print('%s already exists' % mytable)
else:
print('Creating table: %s' % mytable)
try:
# Copy structure and no data (1=2 is false)
cur.execute("""
CREATE TABLE {0} AS SELECT * FROM {1} WHERE 1=2;
""".format(my_table, my_table+'_temp'))
con.commit()
print('Table created successfully')
except psycopg2.DatabaseError as e:
if con:
con.rollback()
print('Error %s' % e)
sys.exit(1)
con.close()
def copy_csv_to_table(path_2_csv='',
my_table=''):
"""
Use the PostgreSQL COPY command to bulk-copy the CSVs into the newly created table
"""
# Connect
con = psycopg2.connect(database=mydatabase, user=myusername, password=mypassword)
cur = con.cursor()
if con:
print('Connected: %s' % con)
else:
print('Connection lost')
sys.exit(1)
copy_sql = """
COPY %s FROM stdin DELIMITERS '~' CSV;
""" % my_table
counter = 0
start_time = time.time()
for i in os.listdir(path_2_csv):
if i.endswith(".csv") & i.startswith("100_pct"):
print("Uploading %s to %s" % (i, mytable))
with open(os.path.join(path_2_csv, i), 'r') as f:
cur.copy_expert(sql=copy_sql, file=f)
con.commit()
counter += 1
print("Successfully uploaded %d CSVs" % counter)
current_speed = ((time.time()-start_time)/60)/counter
print("Average speed is %.2f minutes per database" % current_speed)
con.close()
end_time = time.time()
print("Total duration of COPY: %.2f minutes" % (end_time - start_time)/60)
def sql_query_to_csv(my_table='',
csv_out=''):
"""
Submit query to created PostgreSQL database and output results to a CSV
"""
# Connect
con = psycopg2.connect(database=mydatabase, user=myusername, password=mypassword)
cur = con.cursor()
if con:
print('Connected: %s' % con)
else:
print('Connection lost')
sys.exit(1)
start_time = time.time()
my_query = """
SELECT
SUM("A"),
SUM("B"),
SUM("C"),
COUNT(1) AS "D",
EXTRACT(YEAR FROM "PURCHASE_DATE"::text::date) AS "YEAR",
EXTRACT(MONTH FROM "PURCHASE_DATE"::text::date) AS "MONTH",
.. 20 more columns ...
FROM {0}
GROUP BY
.. 20 columns ...
""".format(my_table)
start_time = time.time()
output_query = "COPY ({0}) TO STDOUT WITH CSV HEADER".format(my_query)
with open(csv_out, 'w') as f:
cur.copy_expert(output_query, f)
print("Successfully submitted results to: %s" % csv_out)
con.close()
end_time = time.time()
print("Total duration of Query: %.2f minutes" % (end_time - start_time)/60)
bcpiscopy: postgresql.org/docs/current/static/sql-copy.html (which processes text file located on the server). Another alternative is pgloader: pgloader.io