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