Python的并行求和例子

import csv
import sys
import logging
import argparse
import numpy as np
import multiprocessing
import time

# configure logging
logger = logging.getLogger("example")

handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter(
    ‘%(asctime)s %(levelname)s %(name)s: %(message)s‘))

logger.addHandler(handler)
logger.setLevel(logging.DEBUG)

def H(n, z):
    return (n-z) >= 0

def evaluate(args, start, end):
    ‘‘‘handle range[start, end)‘‘‘
    logger.info("Started %d to %d" %(start, end))
    expReader = open(‘train_exp.csv‘,‘r‘)
    expReader.readline()
    for i in range(start):
        expReader.readline()
    predFile = open(args.predict)
    for i in range(start+1):
        predFile.readline()
    predReader = csv.reader(predFile, delimiter=‘,‘)
    squareErrorSum = 0
    totalLines = end - start
    for i, row in enumerate(predReader):
        if i == totalLines:
            logger.info("Completed %d to %d" %(start, end))
            break
        expId, exp = expReader.readline().strip().split(‘,‘)
        exp = float(exp)
        predId = row[0]
        row = np.array(row, dtype=‘float‘)
        assert expId == predId
        lineSum = 0
        for j in xrange(1,71):
            n = j - 1
            squareErrorSum += (row[j]-H(n,exp))**2
            lineSum += (row[j]-H(n,exp))**2
    logger.info(‘SquareErrorSum %d to %d: %f‘ %(start, end, squareErrorSum))
    return squareErrorSum

def fileCmp(args):
    ‘‘‘check number of lines in two files are same‘‘‘
    for count, line in enumerate(open(‘train_exp.csv‘)):
        pass
    expLines = count + 1 - 1 #discare header
    for count, line in enumerate(open(args.predict)):
        pass
    predictLines = count + 1 - 1
    print ‘Lines(exp, predict):‘, expLines, predictLines
    assert expLines == predictLines
    evaluate.Lines = expLines
    
if __name__ == "__main__":
    # set up logger
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(‘--predict‘, 
                        help=("path to an predict probability file, this will "
                              "predict_changeTimePeriod.csv"))
    args = parser.parse_args()
    fileCmp(args)
    pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
    result = []
    blocks = multiprocessing.cpu_count()
    linesABlock = evaluate.Lines / blocks
    for i in xrange(blocks-1):
        result.append(pool.apply_async(evaluate, (args, i*linesABlock, (i+1)*linesABlock)))
    result.append(pool.apply_async(evaluate, (args, (i+1)*linesABlock, evaluate.Lines+1)))
    pool.close()
    pool.join()
    result = [res.get() for res in result]
    print result
    print ‘evaluate.Lines‘, evaluate.Lines
    score = sum(result) / (70*evaluate.Lines)
    print "score:", score