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