I'm in the process of filtering some very(!) large files (100gb+): I can't download files with a lower granularity. This is a followup from this question.
The problem is as follows: I need to filter large files that look like the following (3b+ rows).
TIC, Date, Time, Bid, Offer AAPL, 20090901, 09:45, 145, 145.5 AAPL, 20090902, 09:45, 145, 145.5 AAPL, 20090903, 09:45, 145, 145.5
I filter based on TICKER+DATE combinations found in an external file. I have, on average, ~ 1200 dates of interest per firm for ~ 700 firms. The large file contains all dates for the firms of interest, for which I want to extract only a few dates of interest. The big files are split by month (2013-01, 2013-02 etc.).
# Example firm_dates_of_interest file. AAPL, 20090902 AAPL, 20090903
A few changes were made since the previous post:
- Extract the filtering process into a generator function
- Extract chunks from this iterator
I'm currently at 4 minutes of processing time for 30 million rows (1% of the file); I tested a few files and it works properly. However, with about 3 billion rows per file, that puts it at ~8 hours for one 120gb file. Seeing as I have about twelve files, I'm very curious whether I can get significant performance improvements by doing things differently. Example files can be found here (sourcefile, 3gb zipped) and here (firmdates).
Any tips are greatly appreciated.
import os
import datetime
import csv
import cProfile
import re
ROOT_DIR = "H:/ROOT_DIR/"
SOURCE_FILES = os.path.join(ROOT_DIR, '15. Speedtest')
EXPORT_DIR = ROOT_DIR
DATES_FILE = os.path.join(ROOT_DIR, "10. Dates of interest/firm_date_of_interest_withSP.csv")
# Build the original date dict
# For example:
# d['AAPL'] is a list with ['20140901', '20140902', '20140901']
with open(DATES_FILE, "r") as csvfile:
d = {}
reader = csv.reader(csvfile)
reader.next()
for line in reader:
firm = line[1]
date = line[2]
if firm in d.keys():
d[firm].append(date)
else:
d[firm] = [date]
def filter_lines(filename, d):
""" Given a dictionary with key, value a Ticker and dates_of_interest, yield
only the filtered rows with those ticker / dates pairs. """
with open(filename, "rb") as csvfile:
datareader = csv.reader(csvfile)
for row in datareader:
try:
if row[1] in d[row[0]]:
yield row
except KeyError:
continue
except IndexError:
continue
def get_chunk(iterable, chunk_size):
""" Given an iterable and chunk_size, return chunks of chunk_size"""
result = []
for item in iterable:
result.append(item)
if len(result) == chunk_size:
yield tuple(result)
result = []
if len(result) > 0:
yield tuple(result)
def main():
start = datetime.datetime.now()
for root, dir, files in os.walk(SOURCE_FILES):
for i, file in enumerate(files):
basename = os.path.splitext(file)[0]
source_filepath = os.path.join(root, file)
# Annotate files with 'DONE' after succesful processing: skip those
if re.search("DONE", basename):
continue
startfile = datetime.datetime.now()
iterable = filter_lines(source_filepath, d)
for num_saved, chunk in enumerate(get_chunk(iterable, 3000000)):
output_filename = os.path.join(EXPORT_DIR, basename+' EXTRACT' + str(num_saved) + '.csv')
with open(output_filename, 'wb') as csvfile:
writer = csv.writer(csvfile, quoting=csv.QUOTE_NONNUMERIC)
for line in chunk:
writer.writerow(line)
file_elapsed = datetime.datetime.now() - startfile
print("Took me %d seconds for 3000000 extracted rows.." % file_elapsed.seconds)
new_filename = os.path.join(root, os.path.splitext(file)[0]+'- DONE.csv')
os.rename(source_filepath, new_filename)
elapsed = datetime.datetime.now() - start
print("Took me %d seconds..." % elapsed.seconds)
if __name__ == "__main__":
main()