-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathgetLoadStats2.py
executable file
·213 lines (180 loc) · 6.98 KB
/
getLoadStats2.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
#!/usr/bin/python
from pyspark import SparkContext
import sys
#sys.path.append("../")
import os
import logging
import logging.handlers
import resource
import time
import subprocess, threading
from collections import defaultdict
from os.path import realpath
import cProfile
import shutil
import math
from pyspark.accumulators import AccumulatorParam
from helper import parseRecord, getCDFList, FIELD, HO_TYPE, PROC_ID, STATUS, getLoad, getNeighborGradients
class VectorAccumulatorParam(AccumulatorParam):
def zero(self, value):
return [0.0] * len(value)
def addInPlace(self, val1, val2):
for i in xrange(len(val1)):
val1[i] += val2[i]
return val1
def resetDirectories(subDirs,input_dir):
for d in subDirs:
for fname in os.listdir(input_dir+d):
shutil.move(input_dir + d + fname, input_dir + fname)
os.rmdir(input_dir + d)
def getAccumLoad(x):
global timeLoads
global intervals #sorted list
global intervalBoundary #this is a pair specifying the start and end time indices of the interval
global bs2imsi2wasActivePrevTime
bs = x[0]
imsi2time2records = x[1]
#times above correspond to the timeLoads array
#the idea here is decide whether imsi should be added to the time intervals or not
for imsi in imsi2time2records:
v = [0]*len(intervals)
if bs in bs2imsi2wasActivePrevTime and imsi in bs2imsi2wasActivePrevTime[bs]:
wasActiveInPrevTime = bs2imsi2wasActivePrevTime[bs][imsi]
else:
wasActiveInPrevTime = False
for i in range(intervalBoundary[0],intervalBoundary[1]+1):
t = intervals[i]
if t in imsi2time2records[imsi]:
v[i] = 1
#recs = sorted(imsi2time2records[imsi][t])
procID = imsi2time2records[imsi][t][1]
if procID==PROC_ID.CTX_RELEASE_REQ or procID==PROC_ID.DEL_BEARER_REQ:
wasActiveInPrevTime = False
else:
wasActiveInPrevTime = True
else:
if (wasActiveInPrevTime):
v[i] = 1
#v is formed; now add v to timeLoads
timeLoads += v
#time to write to a file
#prevTimeFile.write(bs + " " + imsi + " " + str(wasActiveInPrevTime) + "\n")
bs2imsi2wasActivePrevTime[bs][imsi] = wasActiveInPrevTime
def generateBS2Data(line):
fields = line.split(";")
curBS = fields[FIELD.CUR_CELL-1][0:13]
imsi = fields[FIELD.IMSI-1]
startTime = int(fields[FIELD.START_TIME-1])
procID = int(fields[FIELD.PROC_ID-1])
global intervals
T = 0
for i in range(1,len(intervals)):
t = intervals[i]
if startTime <= t:
T = t
break
assert(T!=0)
#return (curBS,{imsi: {T: [(startTime,procID)]}})
return (curBS,{imsi: {T: (startTime,procID)}})
def filterData(line):
fields = line.split(";")
time = int(fields[FIELD.START_TIME-1])
curBS = fields[FIELD.CUR_CELL-1][0:13]
return time < max(intervals) and len(curBS)>0
def reduceBS2IMSI2Data(x,y):
#x,y are dictionaries of dictionaries; merge them
res = x
for k in y:
if k in res:
y2 = y[k]
for c in y2:
if c in res[k]:
#the '+' operator append the two lists; python construct
#res[k][c] = res[k][c] + y2[c]
#just need to keep the later time
if y2[c][0] >= res[k][c][0]:
res[k][c] = y2[c]
else:
res[k][c] = y2[c]
else:
res[k] = y[k]
return res
if __name__ == "__main__":
if len(sys.argv) < 4:
print >> sys.stderr, "Usage: getLoad <numCores> <interval (min)> <file directory> <file partition size>"
exit(-1)
sys.stdout = open('o.txt', 'w')
numCores = sys.argv[1]
loadInterval = int(sys.argv[2])
input_dir = sys.argv[3]
filePartitionSize = int(sys.argv[4])
if filePartitionSize%loadInterval != 0:
print >> sys.stderr, "file partition size has to be divisible by the load interval. Exiting"
exit(-1)
inputFiles = []
for fname in os.listdir(input_dir):
if fname.find('MMEpcmd') >= 0:
inputFiles.append(fname)
inputFiles = sorted(inputFiles)
startTime = (int(inputFiles[0].split(".")[1][0:2]) + 4)*60*60*1000 + \
int(inputFiles[0].split(".")[1][3:])*60*1000 #start time past utc midnight in millisec
endTime = (int(inputFiles[-1].split(".")[1][0:2]) + 4)*60*60*1000 + \
(int(inputFiles[-1].split(".")[1][3:])+ 1)*60*1000 #end time past utc midnight in millisec
global intervals
intervals = []
for t in range(startTime,endTime+1,loadInterval*60*1000):
intervals.append(t)
intervals = sorted(intervals)
numPartitions = math.ceil(len(inputFiles)/float(filePartitionSize))
dirTimeBoundaries = []
step = int(math.ceil((len(intervals)-1)/numPartitions))
for i in range(step,len(intervals),step):
dirTimeBoundaries.append(intervals[i])
subDirs = []
subDirNum = 0
subFileCount = 0
subDir = str(subDirNum) + "/"
os.makedirs(input_dir + subDir)
for i in range(len(inputFiles)):
f = inputFiles[i]
if subFileCount==filePartitionSize:
subDirs.append(subDir)
subDirNum += 1
subFileCount = 0
subDir = str(subDirNum) + "/"
os.makedirs(input_dir + subDir)
shutil.move(input_dir + f,input_dir + subDir)
subFileCount += 1
if subFileCount==filePartitionSize:
subDirs.append(subDir)
sc = SparkContext("local[" + numCores + "]" , "job", pyFiles=[realpath('helper.py')])
timeLoads = sc.accumulator([0]*len(intervals), VectorAccumulatorParam())
bs2imsi2wasActivePrevTime = defaultdict(lambda: defaultdict())
prev_idx = 0
for i in range(len(subDirs)):
d = subDirs[i]
end_idx = intervals.index(dirTimeBoundaries[i])
intervalBoundary = (prev_idx+1,end_idx) #both indexes are included
prev_idx = end_idx
bs2data = sc.textFile(input_dir + d + '*.gz').filter(filterData).map(generateBS2Data).reduceByKey(reduceBS2IMSI2Data)
bs2data.foreach(getAccumLoad)
print len(bs2imsi2wasActivePrevTime)
resetDirectories(subDirs,input_dir)
sys.exit()
resetDirectories(subDirs,input_dir)
sys.exit()
'''
a = bs2data.first()
for imsi in a[1]:
print "imsi: " + imsi
print "--------------"
for t in sorted(a[1][imsi]):
print str(t) + " " + str(a[1][imsi][t])
#print "---------------"
#for r in sorted(a[1][imsi][t]):
# print r
sys.exit()
'''
mean = [float(x)/numBS for x in timeLoads.value]
print mean
resetDirectories(subDirs,input_dir)