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