spark.py

#!/usr/bin/pypy 
# -*- coding: utf-8 -*-

import logging
import os

logging.basicConfig(level=logging.INFO,format=u'%(asctime)s [%(levelname)s] (%(name)s) {%(pathname)s:%(lineno)d@%(funcName)s} - %(message)s')

if __name__ == '__main__':
    root_dir = os.path.abspath(os.path.dirname(__file__))
    
    dependency = os.path.join(root_dir,'wheels')
    if not os.path.exists(dependency):
        os.makedirs(dependency)
    
    import pip
    pip.main(['wheel','-w',dependency,'-r','requirements.txt'])
    
    pyfiles = set()
    for child in os.listdir(dependency):
        full = os.path.join(dependency,child)
        if not full.endswith('.zip'):
            target = '%s.zip' % full
            os.rename(full,target) 
            full = target
        pyfiles.add(full)
    
    def find_py(root):
        found = set()
        for child in os.listdir(root):
            path = os.path.join(root,child)
            if child.endswith('.py'):
                found.add(path)
            elif os.path.isdir(path):
                found.update(find_py(path))
        return found
    
    pyfiles.update(find_py(root_dir))
    
    from pyspark import SparkContext
    context = SparkContext()
    
    map(lambda x:context.addPyFile(x),pyfiles)
    
    rdd = context.parallelize([i for i in range(100)])
    def run(i):
        import pymongo
        from toolkit import IO
        
        return i

    logging.info(rdd.map(run).collect())
    context.stop()