hiplab/data-transport
hiplab
/
data-transport
mirror of https://github.com/lnyemba/data-transport
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

bug fix: logging with ETL

This commit is contained in:
Steve Nyemba 2022-05-06 14:30:59 -05:00
parent 899339f11f
commit 67cb7de861
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
transport/etl.py
View File

@ -96,7 +96,7 @@ class ETL (Process):
def __init__(self,**_args): def __init__(self,**_args):
super().__init__() super().__init__()
self.name = _args['id'] self.name = _args['id'] if 'id' in _args else 'UNREGISTERED'
if 'provider' not in _args['source'] : if 'provider' not in _args['source'] :
#@deprecate #@deprecate
self.reader = transport.factory.instance(**_args['source']) self.reader = transport.factory.instance(**_args['source'])
@ -126,7 +126,7 @@ class ETL (Process):
# idf = idf.replace({np.nan: None}, inplace = True) # idf = idf.replace({np.nan: None}, inplace = True)
idf.columns = [str(name).replace("b'",'').replace("'","").strip() for name in idf.columns.tolist()] idf.columns = [str(name).replace("b'",'').replace("'","").strip() for name in idf.columns.tolist()]
ETL.logger.info(rows=idf.shape[0],cols=idf.shape[1],jobs=self.JOB_COUNT) # ETL.logger.info(rows=idf.shape[0],cols=idf.shape[1],jobs=self.JOB_COUNT)
# #
# writing the data to a designated data source # writing the data to a designated data source
@ -134,7 +134,7 @@ class ETL (Process):
try: try:
ETL.logger.info(module='write',action='partitioning') # ETL.logger.info(module='write',action='partitioning')
rows = np.array_split(np.arange(0,idf.shape[0]),self.JOB_COUNT) rows = np.array_split(np.arange(0,idf.shape[0]),self.JOB_COUNT)
# #
@ -149,7 +149,7 @@ class ETL (Process):
self.jobs.append(proc) self.jobs.append(proc)
proc.start() proc.start()
ETL.logger.info(module='write',action='working',segment=str(id),table=self.name,rows=segment.shape[0]) # ETL.logger.info(module='write',action='working',segment=str(id),table=self.name,rows=segment.shape[0])
# while poc : # while poc :
# proc = [job for job in proc if job.is_alive()] # proc = [job for job in proc if job.is_alive()]
# time.sleep(1) # time.sleep(1)