· 9 years ago · Nov 27, 2016, 06:40 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 return type
126 try:
127 datetime.strptime(data, '%d-%m-%Y %H:%M:%S')
128 type = ' DATETIME,'
129 return type
130 except ValueError as verr:
131 pass # data does not contain anything convertible to datetime
132 try:
133 datetime.strptime(data, '%Y-%m-%d %H:%M:%S')
134 type = ' DATETIME,'
135 return type
136 except ValueError as verr:
137 pass # data does not contain anything convertible to datetime
138
139 try:
140 float(data)
141 #type = ' FLOAT(22,2),'
142 type = ' varchar(50),'
143 return type
144 except ValueError as verr:
145 pass # data does not contain anything convertible to float
146
147 try:
148 int(data)
149 type = ' BIGINT,'
150 return type
151 except ValueError as verr:
152 pass # data does not contain anything convertible to int
153
154 return type
155
156def getName(filepath, code):
157 with open(filepath,'r') as f:
158 reader=csv.reader(f,delimiter='\t')
159 for row in reader:
160 if row[0]==code:
161 return row[1]
162 return ""
163
164#1. Check Source File
165def checkSourceFile():
166 printAndLog('************1.Check Source File STARTED************')
167 #a) Check file availability in designated location
168 if ifExists(filepath) != True:
169 sys.exit(-1)
170
171 #b) Check if the file is readable
172 if ifReadable(filepath) != True:
173 sys.exit(-1)
174
175 #c) Read file type and meta data from meta data store
176 if fileType(filepath) != '.zip':
177 printAndLog(filepath+ ' is not a zip file')
178 sys.exit(-1)
179 printAndLog(filepath+ ' is a zip file')
180
181 #d) Decompress the file into temp folder
182 if decompress(filepath)!= True:
183 sys.exit(-1)
184
185 #e) Read the manifest file and check the artifacts as mentioned in manifest file
186 if readManifest() != True:
187 sys.exit(-1)
188 printAndLog('************1.Check Source File ENDED************')
189 return True
190
191#2. Create Meta Data
192def createMataData():
193 filepath = LANDING+'/temp'
194 headerFileName = 'column_headers.tsv'
195 dataFileName = dataFile
196 tableName = 'adobe_column_header2'
197 dbName = 'adobe2'
198 createTable = 'create table IF NOT EXISTS '+tableName+' ('
199 printAndLog('************2.Create Meta Data STARTED************')
200 printAndLog("Landing path: "+filepath)
201 printAndLog("headerFileName: "+headerFileName)
202 printAndLog("dataFileName: "+dataFileName)
203 #a) Check if column_header file is available
204 if ifExists(filepath+'/'+headerFileName) != True:
205 sys.exit(-1)
206 #b) Iterate through fields and treat each of them as columns
207 with open(filepath+'/'+dataFileName,'r') as f:
208 reader=csv.reader(f,delimiter='\t')
209 i=0
210 for datarow in reader:
211 if i == 0:
212 break
213 with open(filepath+'/'+headerFileName,'r') as f:
214 reader=csv.reader(f,delimiter='\t')
215 for row in reader:
216 i = 0
217 for y in row:
218 if "videoadinpod" == y.strip():
219 createTable = createTable + y + ' TEXT, '
220 else:
221 createTable = createTable + y + identifyType(datarow[i])
222 i = i + 1
223 createTable = createTable[0:len(createTable)-1]
224 createTable = createTable + ");"
225 # Open database connection ( If database is not created don't give dbname)
226 db = getDBConnection()
227 cursor = db.cursor()
228 # For creating db
229 cursor.execute("create database IF NOT EXISTS "+dbName)
230 cursor.execute("use "+dbName)
231 # create table
232 cursor.execute("drop table IF EXISTS "+tableName)
233 cursor.execute(createTable)
234 printAndLog("table created")
235 # Commit your changes in the database
236 db.commit()
237 # disconnect from server
238 db.close()
239 printAndLog('************2.Create Meta Data ENDED************')
240 return True
241
242#3. Convert File
243def convertFile():
244 printAndLog('************3.convertFile STARTED************')
245 filepath = LANDING+'/temp'
246 headerFileName = 'column_headers.tsv'
247 dataFileName = dataFile
248 languagesFileName= 'languages.tsv'
249 browserFileName= 'browser.tsv'
250 delim = ','
251 if ifExists(filepath+'/'+headerFileName) != True:
252 sys.exit(-1)
253 with open(filepath+'/'+dataFile[:-3]+'csv','w') as df:
254 with open(filepath+'/'+headerFileName,'r') as f:
255 reader=csv.reader(f,delimiter='\t')
256 for row in reader:
257 for y in row:
258 y=y.strip().replace(',',' ')
259 df.write(y+delim)
260 df.write('\n')
261 with open(filepath+'/'+dataFileName,'r') as f:
262 reader=csv.reader(f,delimiter='\t')
263 for row in reader:
264 i=0
265 for y in row:
266 if i==0:
267 df.write(getName(filepath+'/'+languagesFileName,y)+delim)
268 if i==1:
269 df.write(getName(filepath+'/'+browserFileName,y)+delim)
270 if i>1:
271 df.write(y+delim)
272 i = i + 1
273 df.write('\n')
274 df.close();
275 printAndLog('************3.convertFile ENDED************')
276 return True
277
278#4. Transfer Files
279def transferFile():
280 printAndLog('************4. Transfer Files STARTED************')
281 filepath = LANDING+'/temp'
282 #a) Check if converted file is available
283 if ifExists(filepath+'/'+dataFile[:-3]+'csv') != True:
284 sys.exit(-1)
285
286 #b) Read HDFS target location from meta data table
287
288 #c) Copy the file(s) in target location in HDFS
289
290 printAndLog('************4. Transfer Files ENDED************')
291 return True
292
293#5. Validate File
294def validateFile():
295 printAndLog('************5. Validate File STARTED************')
296 #a) Check meta data table to retrieve validations for each column
297
298 #b) Read file from HDFS
299
300 #c) Perform validations
301
302 #d) Store errors in separate file retrieved from meta data table
303
304 #e) Store results in memory
305 printAndLog('************5. Validate File ENDED************')
306
307
308#6. Transform Reference Data
309def transformData():
310 printAndLog('************6. Transform Reference Data STARTED************')
311 #a) Retrieve columns and file names for code replacement from meta data table
312
313 printAndLog('************6. Transform Reference Data ENDED************')
314
315
316
317#returns absolute paths of files in a folder
318def getFilepaths(folderpath):
319 filepaths = [] # to store all of the full filepaths.
320 # Walk
321 for root, directories, files in os.walk(folderpath):
322 for filename in files:
323 # Join the two strings in order to form the full filepath.
324 filepaths.append(os.path.join(root, filename))
325 return filepaths
326
327#creates a table named adobe_url.
328#In this table it should have the following columns id, name, url.
329#You should populate the table with the name of file and the url of the file location
330#e.g c:/files/landing/temp/column_header.tsv
331def createMetaDataTable():
332 printAndLog('************createMetaDataTable STARTED************')
333 tableName = 'adobe_url'
334 filepath = LANDING+'/temp'
335 dbName = 'adobe2'
336 createTable = 'create table IF NOT EXISTS '+tableName+' (ID int NOT NULL AUTO_INCREMENT,'
337 createTable = createTable + 'fileName varchar(255) NOT NULL, URL varchar(255) NOT NULL, hdfs_url varchar(255), PRIMARY KEY (ID));'
338 db = getDBConnection()
339 cursor = db.cursor()
340 # For creating db
341 cursor.execute("create database IF NOT EXISTS "+dbName)
342 cursor.execute("use "+dbName)
343 # create table
344 cursor.execute("drop table IF EXISTS "+tableName)
345 cursor.execute(createTable)
346 printAndLog("table created: "+tableName)
347 # reading files names
348 paths = getFilepaths(filepath)
349 for fp in paths:
350 head, tail = os.path.split(fp)
351 insertStatement = "INSERT INTO "+tableName+"(fileName,URL,hdfs_url) values ('"+tail+"','"+fp+"','"+(hdfs_rawdatalanding+"/"+tail)+"')"
352 cursor.execute(insertStatement)
353 # Commit your changes in the database
354 db.commit()
355 # disconnect from server
356 db.close()
357 printAndLog('************createMetaDataTable ENDED************')
358
359
360if checkSourceFile():
361 printAndLog('=====>1.Check Source File completed successfully')
362 if createMataData():
363 printAndLog('=====>2.Create Meta Data completed successfully')
364 else:
365 printAndLog('=====>2.Create Meta Data not completed successfully')
366 if convertFile():
367 printAndLog('=====>3.convertFile completed successfully')
368 createMetaDataTable()
369 if transferFile():
370 printAndLog('=====>4. Transfer Files completed successfully')
371 validateFile()
372 transformData()
373
374 else:
375 printAndLog('=====>4. Transfer Files not completed successfully')
376 else:
377 printAndLog('=====>3.convertFile not completed successfully')
378else:
379 printAndLog('=====>1.Check Source File not completed successfully')