· 9 years ago · Nov 27, 2016, 09:18 PM
1#Import Libraries
2import mysql.connector
3import csv
4from datetime import datetime
5
6import os
7import zipfile
8import gzip
9import glob
10import os.path
11import tarfile
12
13LANDING = 'C:/files/landing'
14filepath = LANDING+'/we.Finance-DataFeed_example.zip'
15manifestFile= 'geo1xxlonfsi2142_2016-10-19.txt'
16lookupFile=''
17dataFile='01-geo1xxlonfsi2142_2016-10-19.tsv'
18hdfs_base = '/adobeanalytics/data'
19hdfs_rawdatalanding='/adobeanalytics/data/rawdatalanding'
20hdfs_cleandata='/adobeanalytics/data/cleandata'
21hdfs_validateddata='/adobeanalytics/data/validateddata'
22hdfs_validedata='/adobeanalytics/data/validedata'
23class myGzipFile(gzip.GzipFile):
24 def __enter__(self):
25 if self.fileobj is None:
26 raise ValueError("I/O operation on closed GzipFile object")
27 return self
28
29 def __exit__(self, *args):
30 self.close()
31
32def printAndLog(msg):
33 f=open(LANDING+"/log.txt", "a+")
34 print(msg)
35 now = datetime.now()
36 f.write(str(now)+": "+msg+"\n")
37
38#function to connect to database
39def getDBConnection():
40 try:
41 db = mysql.connector.connect( user="root", password="admin", database="", host="localhost")
42 printAndLog("Connection to mySQL Successful.")
43 return db
44 except Exception as e:
45 printAndLog(e)
46 sys.exit(-1)
47
48#Return True if path is an existing regular file
49def ifExists(filepath):
50 try:
51 open(filepath,'r')
52 printAndLog('File '+filepath+' exists')
53 return True
54 except Exception as e:
55 printAndLog(e)
56 sys.exit(-1)
57#to Check if the file is readable
58def ifReadable(filepath):
59 try:
60 with open(filepath,'r') as f:
61 printAndLog('File '+filepath+' is readable')
62 return True
63 except Exception as e:
64 printAndLog(e)
65 printAndLog('File '+filepath+' is not readable')
66 sys.exit(-1)
67#to return filepath
68def fileType(filepath):
69 ext = os.path.splitext(filepath)[-1].lower()
70 return ext
71
72#to decompress
73def decompress(filepath):
74 try:
75 zip_ref = zipfile.ZipFile(filepath, 'r')
76 zip_ref.extractall(LANDING+'/temp/')
77 zip_ref.close()
78 printAndLog(filepath+ ' decompressed to '+LANDING+'/temp')
79 return True
80 except Exception as e:
81 printAndLog(filepath+' '+ str(e))
82 sys.exit(-1)
83
84def readManifest():
85 printAndLog('Reading manifest: '+manifestFile)
86 path = LANDING+'/temp/'+manifestFile
87 if ifExists(path) != True:
88 sys.exit(-1)
89 if ifReadable(path) != True:
90 sys.exit(-1)
91 with open(path,'r') as f:
92 for line in f:
93 line = line.rstrip('\n')
94 if len(line) > 0:
95 tokens=line.split(":")
96 tokens[1]=tokens[1].strip()
97 if tokens[0]=="Data-File":
98 dataFile = tokens[1]
99 if tokens[0]=="Lookup-File":
100 lookupFile = tokens[1]
101 source_dir = LANDING+'/temp/'
102 dest_dir = source_dir
103 for src_name in glob.glob(os.path.join(source_dir, '*.gz')):
104 base = os.path.basename(src_name)
105 dest_name = os.path.join(dest_dir, base[:-3])
106 with myGzipFile(src_name, 'rb') as infile:
107 with open(dest_name, 'wb') as outfile:
108 for line in infile:
109 outfile.write(line)
110 try:
111 dataFile = LANDING+'/temp/'+dataFile[:-3]
112 lookupFile = LANDING+'/temp/'+lookupFile[:-3]
113 tar = tarfile.open(lookupFile)
114 tar.extractall(LANDING+'/temp')
115 tar.close()
116 except Exception as e:
117 printAndLog(e)
118 sys.exit(-1)
119
120 return True
121
122#to identify the data type of column
123def identifyType(data):
124 type = ' varchar(50),'
125 try:
126 datetime.strptime(data, '%d-%m-%Y %H:%M:%S')
127 type = ' DATETIME,'
128 return type
129 except ValueError as verr:
130 pass # data does not contain anything convertible to datetime
131 try:
132 datetime.strptime(data, '%Y-%m-%d %H:%M:%S')
133 type = ' DATETIME,'
134 return type
135 except ValueError as verr:
136 pass # data does not contain anything convertible to datetime
137 try:
138 int(data)
139 type = ' INTEGER,'
140 return type
141 except ValueError as verr:
142 pass # data does not contain anything convertible to int
143 try:
144 float(data)
145 type = ' FLOAT('+str(len(data))+',2),'
146 return type
147 except ValueError as verr:
148 pass # data does not contain anything convertible to float
149 return type
150
151def getName(filepath, code):
152 with open(filepath,'r') as f:
153 reader=csv.reader(f,delimiter='\t')
154 for row in reader:
155 if row[0]==code:
156 return row[1]
157 return ""
158
159#1. Check Source File
160def checkSourceFile():
161 printAndLog('************1.Check Source File STARTED************')
162 #a) Check file availability in designated location
163 if ifExists(filepath) != True:
164 sys.exit(-1)
165
166 #b) Check if the file is readable
167 if ifReadable(filepath) != True:
168 sys.exit(-1)
169
170 #c) Read file type and meta data from meta data store
171 if fileType(filepath) != '.zip':
172 printAndLog(filepath+ ' is not a zip file')
173 sys.exit(-1)
174 printAndLog(filepath+ ' is a zip file')
175
176 #d) Decompress the file into temp folder
177 if decompress(filepath)!= True:
178 sys.exit(-1)
179
180 #e) Read the manifest file and check the artifacts as mentioned in manifest file
181 if readManifest() != True:
182 sys.exit(-1)
183 printAndLog('************1.Check Source File ENDED************')
184 return True
185
186#2. Create Meta Data
187def createMataData():
188 filepath = LANDING+'/temp'
189 headerFileName = 'column_headers.tsv'
190 dataFileName = dataFile
191 tableName = 'adobe_column_header2'
192 dbName = 'adobe2'
193 createTable = 'create table IF NOT EXISTS '+tableName+' ('
194 printAndLog('************2.Create Meta Data STARTED************')
195 printAndLog("Landing path: "+filepath)
196 printAndLog("headerFileName: "+headerFileName)
197 printAndLog("dataFileName: "+dataFileName)
198 #a) Check if column_header file is available
199 if ifExists(filepath+'/'+headerFileName) != True:
200 sys.exit(-1)
201 #b) Iterate through fields and treat each of them as columns
202 with open(filepath+'/'+dataFileName,'r') as f:
203 reader=csv.reader(f,delimiter='\t')
204 i=0
205 for datarow in reader:
206 if i == 0:
207 break
208 with open(filepath+'/'+headerFileName,'r') as f:
209 reader=csv.reader(f,delimiter='\t')
210 for row in reader:
211 i = 0
212 for y in row:
213 if "videoadinpod" == y.strip():
214 createTable = createTable + y + ' TEXT, '
215 else:
216 createTable = createTable + y + identifyType(datarow[i])
217 i = i + 1
218 createTable = createTable[0:len(createTable)-1]
219 createTable = createTable + ");"
220 # Open database connection ( If database is not created don't give dbname)
221 db = getDBConnection()
222 cursor = db.cursor()
223 # For creating db
224 cursor.execute("create database IF NOT EXISTS "+dbName)
225 cursor.execute("use "+dbName)
226 # create table
227 cursor.execute("drop table IF EXISTS "+tableName)
228 cursor.execute(createTable)
229 printAndLog("table created")
230 # Commit your changes in the database
231 db.commit()
232 # disconnect from server
233 db.close()
234 printAndLog('************2.Create Meta Data ENDED************')
235 return True
236
237#3. Convert File
238def convertFile():
239 printAndLog('************3.convertFile STARTED************')
240 filepath = LANDING+'/temp'
241 headerFileName = 'column_headers.tsv'
242 dataFileName = dataFile
243 languagesFileName= 'languages.tsv'
244 browserFileName= 'browser.tsv'
245 delim = ','
246 if ifExists(filepath+'/'+headerFileName) != True:
247 sys.exit(-1)
248 with open(filepath+'/'+dataFile[:-3]+'csv','w') as df:
249 with open(filepath+'/'+headerFileName,'r') as f:
250 reader=csv.reader(f,delimiter='\t')
251 for row in reader:
252 for y in row:
253 y=y.strip().replace(',',' ')
254 df.write(y+delim)
255 df.write('\n')
256 with open(filepath+'/'+dataFileName,'r') as f:
257 reader=csv.reader(f,delimiter='\t')
258 for row in reader:
259 i=0
260 for y in row:
261 if i==0:
262 df.write(getName(filepath+'/'+languagesFileName,y)+delim)
263 if i==1:
264 df.write(getName(filepath+'/'+browserFileName,y)+delim)
265 if i>1:
266 df.write(y+delim)
267 i = i + 1
268 df.write('\n')
269 df.close();
270 printAndLog('************3.convertFile ENDED************')
271 return True
272
273#4. Transfer Files
274def transferFile():
275 printAndLog('************4. Transfer Files STARTED************')
276 filepath = LANDING+'/temp'
277 #a) Check if converted file is available
278 if ifExists(filepath+'/'+dataFile[:-3]+'csv') != True:
279 sys.exit(-1)
280
281 #b) Read HDFS target location from meta data table
282
283 #c) Copy the file(s) in target location in HDFS
284
285 printAndLog('************4. Transfer Files ENDED************')
286 return True
287
288#5. Validate File
289def validateFile():
290 printAndLog('************5. Validate File STARTED************')
291 #a) Check meta data table to retrieve validations for each column
292
293 #b) Read file from HDFS
294
295 #c) Perform validations
296
297 #d) Store errors in separate file retrieved from meta data table
298
299 #e) Store results in memory
300 printAndLog('************5. Validate File ENDED************')
301
302
303#6. Transform Reference Data
304def transformData():
305 printAndLog('************6. Transform Reference Data STARTED************')
306 #a) Retrieve columns and file names for code replacement from meta data table
307
308 printAndLog('************6. Transform Reference Data ENDED************')
309
310
311
312#returns absolute paths of files in a folder
313def getFilepaths(folderpath):
314 filepaths = [] # to store all of the full filepaths.
315 # Walk
316 for root, directories, files in os.walk(folderpath):
317 for filename in files:
318 # Join the two strings in order to form the full filepath.
319 filepaths.append(os.path.join(root, filename))
320 return filepaths
321
322#creates a table named adobe_url.
323#In this table it should have the following columns id, name, url.
324#You should populate the table with the name of file and the url of the file location
325#e.g c:/files/landing/temp/column_header.tsv
326def createMetaDataTable():
327 printAndLog('************createMetaDataTable STARTED************')
328 tableName = 'adobe_url'
329 filepath = LANDING+'/temp'
330 dbName = 'adobe2'
331 createTable = 'create table IF NOT EXISTS '+tableName+' (ID int NOT NULL AUTO_INCREMENT,'
332 createTable = createTable + 'fileName varchar(255) NOT NULL, URL varchar(255) NOT NULL, hdfs_url varchar(255), PRIMARY KEY (ID));'
333 db = getDBConnection()
334 cursor = db.cursor()
335 # For creating db
336 cursor.execute("create database IF NOT EXISTS "+dbName)
337 cursor.execute("use "+dbName)
338 # create table
339 cursor.execute("drop table IF EXISTS "+tableName)
340 cursor.execute(createTable)
341 printAndLog("table created: "+tableName)
342 # reading files names
343 paths = getFilepaths(filepath)
344 for fp in paths:
345 head, tail = os.path.split(fp)
346 insertStatement = "INSERT INTO "+tableName+"(fileName,URL,hdfs_url) values ('"+tail+"','"+(head+"/"+tail)+"','"+(hdfs_rawdatalanding+"/"+tail)+"')"
347 cursor.execute(insertStatement)
348 # Commit your changes in the database
349 db.commit()
350 # disconnect from server
351 db.close()
352 printAndLog('************createMetaDataTable ENDED************')
353
354
355if checkSourceFile():
356 printAndLog('=====>1.Check Source File completed successfully')
357 if createMataData():
358 printAndLog('=====>2.Create Meta Data completed successfully')
359 else:
360 printAndLog('=====>2.Create Meta Data not completed successfully')
361 if convertFile():
362 printAndLog('=====>3.convertFile completed successfully')
363 createMetaDataTable()
364 if transferFile():
365 printAndLog('=====>4. Transfer Files completed successfully')
366 validateFile()
367 transformData()
368
369 else:
370 printAndLog('=====>4. Transfer Files not completed successfully')
371 else:
372 printAndLog('=====>3.convertFile not completed successfully')
373else:
374 printAndLog('=====>1.Check Source File not completed successfully')