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