· 8 years ago · Aug 23, 2018, 10:54 AM
1from __future__ import print_function
2import sys
3import os
4import json
5import re
6import csv
7import logging
8import traceback
9from datetime import datetime
10from subprocess import Popen, PIPE
11
12
13def getDateTime(fmt):
14 ''' Get current date time in the specified format '''
15 current_time = datetime.now()
16 return current_time.strftime(fmt)
17
18
19def initLogging(logFilePath):
20 ''' Initializes Logging '''
21 logFileName = "csvToHive_" + getDateTime(fmt="%Y%m%d%H%M%S")+".log"
22 if not (os.path.exists(logFilePath)):
23 os.makedirs(logFilePath, 0775)
24 global logging
25 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s:%(message)s',
26 datefmt='%m/%d/%Y %I:%M:%S %p', filename=logFilePath+'/'+logFileName)
27
28
29def getSrcFileNames(srcFilePath, regExForFileName):
30 ''' get list of fileNames from srcFilePath using provided regular expression '''
31 status = -1
32 fileNamesList = []
33 if os.path.exists(srcFilePath):
34 status = 0
35 regex = re.compile(regExForFileName, re.IGNORECASE)
36 for root, dirs, files in os.walk(srcFilePath):
37 for File in files:
38 result = regex.findall(File)
39 if result:
40 fileNamesList.append(result[0])
41 return fileNamesList, status
42
43
44def archiveFiles(srcFilePath, srcFileName, targetPath):
45 ''' archive the files to the targetPath and append datetime to the file '''
46 logging.info("Archiving File")
47 if not (os.path.exists(targetPath)):
48 os.makedirs(targetPath, 0770)
49
50 fileNamewoExtn = srcFileName.split(".")[0]
51 fileNameExtnList = srcFileName.split(".")[1:]
52 fileNameExtn = ".".join(fileNameExtnList)
53 destFileName = fileNamewoExtn + "-" + getDateTime(fmt="%Y%m%d%H%M%S") + "." + fileNameExtn
54 strCommand = "mv "" + srcFilePath + "/" + srcFileName + "" "" + targetPath + "/" + destFileName + """
55 logging.info(strCommand)
56 archiveSubProcess = Popen(strCommand, shell=True, stdout=PIPE, stderr=PIPE)
57 archive_err = archiveSubProcess.communicate()[1]
58 return archive_err
59
60
61def remove_quotes(string):
62 ''' This function removes the quotes from the string passed to this function'''
63 return ''.join(c for c in string if c not in ('"', "'"))
64
65
66def remove_alphanumeric(string):
67 ''' This function removes alpha-numeric characters from the string passed to this function '''
68 return ''.join(e for e in string if e.isalnum())
69
70
71def loadCSVFileToHiveTable(csvFilesGeneratedList):
72 status = 0
73 with open(configFile) as json_data:
74 config = json.load(json_data)
75 for csvFile in csvFilesGeneratedList:
76 logging.info("--------------------------------------------")
77 print("n")
78 logging.info("Loading CSV File - %s into Hive Table" % csvFile)
79 print("Loading CSV File", csvFile, "into Hive Table")
80 configFound = False
81 for csv_hive in config["CSVFileToHiveTable"]:
82 if csv_hive['csvSheetName'] in csvFile:
83 csvSheetName = csv_hive['csvSheetName']
84 hiveDB_TableName = csv_hive['hiveDB_TableName']
85 fieldsTerminated_by = csv_hive['fieldsTerminated_by']
86 hiveTableHDFSLocation = csv_hive['hiveTableHDFSLocation']
87 configFound = True
88 break
89
90 if configFound == False:
91 logging.error("No Hive Configurations found for CSV File - %s" % csvFile)
92 print("No Hive Configurations found for CSV File -", csvFile)
93 else:
94 with open(csvFile, 'rb') as fin:
95 csvin = csv.reader(fin)
96 headerRec = next(csvin, [])
97 hiveSchema = ", ".join([col + " string" for col in headerRec])
98 hiveDDL = "CREATE EXTERNAL TABLE IF NOT EXISTS " + hiveDB_TableName + " (" + hiveSchema + " )"
99 hiveDDL += " row format serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde' "
100 # hiveDDL += " row format serde 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' "
101 hiveDDL += " with serdeproperties ('field.delim' = '" + fieldsTerminated_by + "') "
102 hiveDDL += "STORED AS TEXTFILE LOCATION '" + hiveTableHDFSLocation + "' "
103 hiveDDL += "tblproperties ('skip.header.line.count'='1','orc.compress'='SNAPPY');"
104
105 dropCommand = "hive -S -e "DROP TABLE IF EXISTS " + hiveDB_TableName + "";"
106 logging.info("Drop existing Hive Table command - %s" % dropCommand)
107 print("Dropping Hive Table", hiveDB_TableName)
108 Popen(dropCommand, shell=True, stdout=PIPE, stderr=PIPE).communicate()
109
110 createCommand = "hive -S -e "" + hiveDDL + """
111 logging.info("Create Table Command - %s" % createCommand)
112 print("Creating Hive Table", hiveDB_TableName)
113 createSubProcess = Popen(createCommand, shell=True, stdout=PIPE, stderr=PIPE)
114 err = createSubProcess.communicate()[1]
115 retCode = createSubProcess.returncode
116 if retCode != 0:
117 print("Error in creating Hive Table.. Please check logs")
118 logging.error("Error in creating Hive Table..")
119 logging.error(err)
120 print("Data Load Failed for Excel Sheet", csvSheetName)
121 logging.error("Data Load Failed for Excel Sheet - %s" % csvSheetName)
122 status += 1
123 else:
124 loadCommand = "hive -S -e "LOAD DATA LOCAL INPATH '" + csvFile + "' OVERWRITE INTO TABLE "
125 + hiveDB_TableName + ";""
126 logging.info("Load Data into Hive Table command - %s" % loadCommand)
127 loadSubProcess = Popen(loadCommand, shell=True, stdout=PIPE, stderr=PIPE)
128 err = loadSubProcess.communicate()[1]
129 retCode = loadSubProcess.returncode
130 if retCode != 0:
131 print("Error in Loading CSV File into Hive Table.. Please check logs")
132 logging.error("Error in Loading CSV File into Hive Table..")
133 logging.error(err)
134 print("Data Load Failed for Excel Sheet", csvSheetName)
135 logging.error("Data Load Failed for Excel Sheet - %s" % csvSheetName)
136 status += 1
137 else:
138 print("Successfully Loaded CSV File into Hive Table", hiveDB_TableName)
139 logging.info("Successfully Loaded CSV File - %s into Hive Table - %s" % (csvFile, hiveDB_TableName))
140 logging.info("Archiving CSV File - %s archive path - %s" % (csvFile, archiveBasePath))
141 archive_err = archiveFiles("/".join(csvFile.split("/")[:-1]), csvFile.split("/")[-1],
142 archiveBasePath)
143
144 logging.info("--------------------------------------------")
145 return status
146
147
148def processConfig(configFile):
149 exitCode = 0
150 with open(configFile) as json_data:
151 config = json.load(json_data)
152 logFilePath = config["logFilePath"]
153
154 # initialize Logging
155 initLogging(logFilePath)
156 logging.info("PROCESS STARTED")
157
158 for csvFileSource in config["csvFileSource"]:
159 srcFilePath = csvFileSource['sourceFilePath']
160 regExForSrcFileName = csvFileSource['regExForSrcFileName']
161 targetPath = csvFileSource['targetPath']
162 workSheetsToExport = csvFileSource['workSheetsToExport']
163 global archiveBasePath
164 archiveBasePath = csvFileSource["archiveBasePath"]
165
166 if not os.path.exists(targetPath):
167 os.makedirs(targetPath, 0777)
168
169 srcFileNamesList, status = getSrcFileNames(srcFilePath, regExForSrcFileName)
170 if status == 0:
171 if len(srcFileNamesList)!= 0:
172 for srcFileName in srcFileNamesList:
173 logging.info("********************************************")
174 logging.info("Processing File: %s" % (srcFilePath + "/" + srcFileName))
175 logging.info("********************************************")
176
177 process_status, csvFilesGeneratedList = processExcelFile(
178 srcFile=srcFilePath + "/" + srcFileName, targetPath=targetPath,
179 workSheetsToExport=workSheetsToExport)
180
181 # archive files
182 if process_status == 0:
183 logging.info("Successfully Converted Excel File: %s to CSV File(s)" % (
184 srcFilePath + "/" + srcFileName))
185 if archiveBasePath is not None:
186 archive_err = archiveFiles(srcFilePath, srcFileName, archiveBasePath)
187 if not archive_err:
188 logging.info("Archived File: %s to Location: %s" % (
189 srcFilePath + "/" + srcFileName, archiveBasePath))
190 else:
191 logging.error("Failed to Archive File: %s" % (srcFilePath + "/" + srcFileName))
192 exitCode += 1
193 else:
194 logging.info("Skipping Archiving File: %s" % (srcFilePath + "/" + srcFileName))
195 else:
196 logging.error(
197 "Failed to Convert Excel File: %s to CSV File(s)" % (srcFilePath + "/" + srcFileName))
198 logging.error("Skipping Archiving File: %s" % (srcFilePath + "/" + srcFileName))
199 exitCode += 1
200 logging.info("********************************************")
201
202 logging.info("********************************************")
203 logging.info("Loading the generated CSV files into Hive Tables")
204 logging.info("Excel File : %s" % srcFileName)
205 logging.info("Generated CSV Files : %s" % csvFilesGeneratedList)
206 exitCode += loadCSVFileToHiveTable(csvFilesGeneratedList)
207 logging.info("Loading Files Completed")
208
209 else:
210 logging.warn("No File found with RegEx %s in Path %s" % (regExForSrcFileName, srcFilePath))
211 exitCode += 1
212 else:
213 logging.error("Path: %s does not Exist" % (srcFilePath))
214 exitCode += 1
215
216 return exitCode
217
218
219if __name__ == '__main__':
220 if len(sys.argv) != 2:
221 print("Please pass json config file", file=sys.stderr)
222 exit(-1)
223
224 global configFile
225 configFile = str(sys.argv[1])
226
227 exitCode = processConfig(configFile)
228
229 logging.info("Program Exit Code: %s" % exitCode)
230 logging.info("PROCESS ENDED")
231
232 if exitCode != 0:
233 sys.exit(exitCode)
234
235process_status,csvFilesGeneratedList=processExcelFile(srcFile=srcFilePath + "/" + srcFileName, targetPath=targetPath,workSheetsToExport=workSheetsToExport)