pythonビッグデータ分析オペレーティングシステムログ

1つのコード
1、大きいファイルは切り分ける

import os
import os.path
import time
def FileSplit(sourceFile, targetFolder):
    if not os.path.isfile(sourceFile):
        print(sourceFile, ' does not exist.')
        return
    if not os.path.isdir(targetFolder):
        os.mkdir(targetFolder)
    tempData = []
    number = 1000
    fileNum = 1
    linesRead = 0    
    with open(sourceFile, 'r') as srcFile:
        dataLine = srcFile.readline().strip()
        while dataLine:
            for i in range(number):
                tempData.append(dataLine)                
                dataLine = srcFile.readline()                
                if not dataLine:
                    break
            desFile = os.path.join(targetFolder, sourceFile[0:-4] + str(fileNum) + '.txt')
            with open(desFile, 'a+') as f:
                f.writelines(tempData)
            tempData = []
            fileNum = fileNum + 1       
if __name__ == '__main__':
    #sourceFile = input('Input the source file to split:')
    #targetFolder = input('Input the target folder you want to place the split files:')
    sourceFile = 'test.txt'
    targetFolder = 'test'
    FileSplit(sourceFile, targetFolder)

 
2、Mapperコード

import os
import re
import threading
import time
def Map(sourceFile):
    if not os.path.exists(sourceFile):
        print(sourceFile, ' does not exist.')
        return    
    pattern = re.compile(r'[0-9]{1,2}/[0-9]{1,2}/[0-9]{4}')
    result = {}
    with open(sourceFile, 'r') as srcFile:
        for dataLine in srcFile:
            r = pattern.findall(dataLine)
            if r:
                t = result.get(r[0], 0)
                t += 1
                result[r[0]] = t
    desFile = sourceFile[0:-4] + '_map.txt'
    with open(desFile, 'a+') as fp:
        for k, v in result.items():
            fp.write(k + ':' + str(v) + '
')
if __name__ == '__main__':
    desFolder = 'test'
    files = os.listdir(desFolder)
    
    #        ，       
    '''for f in files:
        Map(desFolder + '\\' + f)'''
    
    #     
    def Main(i):
        Map(desFolder + '\\' + files[i])
    fileNumber = len(files)
    for i in range(fileNumber):
        t = threading.Thread(target = Main, args =(i,))
        t.start()

 
3.Reducerコード

import os
def Reduce(sourceFolder, targetFile):
    if not os.path.isdir(sourceFolder):
        print(sourceFolder, ' does not exist.')
        return
    result = {}
    #Deal only with the mapped files
    allFiles = [sourceFolder+'\\'+f for f in os.listdir(sourceFolder) if f.endswith('_map.txt')]
    for f in allFiles:
        with open(f, 'r') as fp:
            for line in fp:
                line = line.strip()
                if not line:
                    continue
                position = line.index(':')
                key = line[0:position]
                value = int(line[position + 1:])
                result[key] = result.get(key,0) + value
    with open(targetFile, 'w') as fp:
        for k,v in result.items():
            fp.write(k + ':' + str(v) + '
')
if __name__ == '__main__':
    Reduce('test', 'test\\result.txt')

 
二運転結果
上記の3つのプログラムを順次実行し、最終結果を得ます.
07/10/2013:4634
07/16/2013:51
08/15/2013:3958
07/11/2013:1
10/09/2013:733
12/11/2013:564
02/12/2014:4102
05/14/2014:737