/threaded-mongo-document-processor

Primary LanguagePython

Threaded Document Processor

The purpose of this repo is to facilitate multi-threaded processing of mongo collections. The parent class ThreadedDocumentProcessor.py should not be changed. This class will handle threading, exceptions, errors, progress messages, and document iteration.

How To Use

Dependencies

  • python3 -m pip install --user pymongo

What to Change

  • Update the query variable in main() of processDocuments.py to specify a query on the collection
  • Update the processDocument() method to specify what the program does to each document in the collection
    • If you return anything from the function, it must be a python dictionary
    • This dictionary will be formatted into json and written to the output.json file

Running the Program

  • python3 processDocuments.py <collection_name> <number_of_threads>

Files

errors.log

  • This file is automatically created when the program is run if any exceptions are caught
  • This is the log file for any exceptions that arise during document processing

output.json

  • This file is automatically created when the program is run
  • This is where the program will write json output, if any

utils.py

  • Utility functions for ThreadedDocumentProcessor.py

ThreadedDocumentProcessor.py

  • Again, this class should not be changed.
  • This class performs the following
    • Makes a connection to mongodb.
    • Gets a cursor to the collection of interest, with a query specified by the user.
    • Creates threads and breaks up the input space (the documents of the specified collection) among the threads.
    • Calls the user-specified processDocument() method from each thread on each document assigned to that thread.
    • Handles any exceptions that arise and prints error messages to the errors.log file.
    • Logs progress messages to the console.
    • Writes any user-specified data to the output.json file.

processDocuments.py

  • Takes command-line input (see Running the Program section) and passes it to the program constructor
  • Defines a processDocument() method to be filled out by the user

In Case of Emergency

  • If you need to stop the program and restart it for any reason, you can read the document_number as the last integer printed in each thread's progress output, and documents_processed_by_this_thread as the numerator of the fraction in the same progress message, and pass those in explicitly in ThreadedDocumentProcessor.py as two additional arguments, in that order, in the run() method.
  • It would look like this: thread = Thread(target=ThreadedDocumentProcessor.iterateDocuments, args=(self, i, <document_number>, <documents_processed_by_this_thread>))
  • This will cause the script to restart where it left off.
  • Note: this happens automatically if an exception is thrown, just not if you kill the script with ctrl+c