|
|
|
@ -5,113 +5,42 @@ NOTE: Plugins are converted to a pipeline, so we apply a pipeline when reading o
|
|
|
|
- upon initialization we will load plugins
|
|
|
|
- upon initialization we will load plugins
|
|
|
|
- on read/write we apply a pipeline (if passed as an argument)
|
|
|
|
- on read/write we apply a pipeline (if passed as an argument)
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
from transport.plugins import PluginLoader
|
|
|
|
from transport.plugins import Plugin, PluginLoader
|
|
|
|
import transport
|
|
|
|
import transport
|
|
|
|
from transport import providers
|
|
|
|
from transport import providers
|
|
|
|
from multiprocessing import Process, RLock
|
|
|
|
from multiprocessing import Process
|
|
|
|
import time
|
|
|
|
import time
|
|
|
|
import types
|
|
|
|
|
|
|
|
from . import registry
|
|
|
|
|
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
import pandas as pd
|
|
|
|
|
|
|
|
import numpy as np
|
|
|
|
|
|
|
|
import os
|
|
|
|
|
|
|
|
import sys
|
|
|
|
|
|
|
|
import itertools
|
|
|
|
|
|
|
|
import json
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import plugin_ix
|
|
|
|
import plugin_ix
|
|
|
|
|
|
|
|
|
|
|
|
class BaseIO :
|
|
|
|
class IO:
|
|
|
|
def __init__(self,**_args):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self._logger = _args['logger'] if 'logger' in _args else None
|
|
|
|
|
|
|
|
self._logTable = 'logs' if 'logTable' not in _args else _args['logTable']
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def setLogger(self,_logger):
|
|
|
|
|
|
|
|
self._logger = _logger
|
|
|
|
|
|
|
|
def log (self,**_args):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if self._logger :
|
|
|
|
|
|
|
|
_date = str(datetime.now())
|
|
|
|
|
|
|
|
_data = dict({'pid':os.getpid(),'date':_date[:10],'time':_date[11:19]},**_args)
|
|
|
|
|
|
|
|
for key in _data :
|
|
|
|
|
|
|
|
if type(_data[key]) == list :
|
|
|
|
|
|
|
|
_data[key] = [_item.__name__ if type(_item).__name__== 'function' else _item for _item in _data[key]]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_data[key] = str(_data[key]) if type(_data[key]) not in [list,dict] else json.dumps(_data[key])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self._logger.write(pd.DataFrame([_data])) #,table=self._logTable)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class IO(BaseIO):
|
|
|
|
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
Base wrapper class for read/write and support for logs
|
|
|
|
Base wrapper class for read/write and support for logs
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
def __init__(self,**_args):
|
|
|
|
def __init__(self,**_args):
|
|
|
|
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
# We need to initialize the logger here ...
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
super().__init__(**_args)
|
|
|
|
|
|
|
|
_agent = _args['agent']
|
|
|
|
_agent = _args['agent']
|
|
|
|
plugins = _args['plugins'] if 'plugins' else None
|
|
|
|
plugins = _args['plugins'] if 'plugins' in _args else None
|
|
|
|
# _logger = _args['logger'] if 'logger' in _args else None
|
|
|
|
|
|
|
|
# self._logger = _logger if not type(_agent) in [IReader,IWriter] else _agent._logger #transport.get.writer(label='logger') #if registry.has('logger') else None
|
|
|
|
|
|
|
|
# if not _logger and hasattr(_agent,'_logger') :
|
|
|
|
|
|
|
|
# self._logger = getattr(_agent,'_logger')
|
|
|
|
|
|
|
|
self._agent = _agent
|
|
|
|
self._agent = _agent
|
|
|
|
_date = _date = str(datetime.now())
|
|
|
|
# self._ixloader = plugin_ix.Loader () #-- must indicate where the plugin registry file is
|
|
|
|
self._ixloader = plugin_ix.Loader (registry=plugin_ix.Registry(folder=transport.registry.REGISTRY_PATH))
|
|
|
|
self._ixloader = plugin_ix.Loader (registry=plugin_ix.Registry(folder=transport.registry.REGISTRY_PATH))
|
|
|
|
|
|
|
|
|
|
|
|
# self._logTable = 'logs' #'_'.join(['logs',_date[:10]+_date[11:19]]).replace(':','').replace('-','_')
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if plugins :
|
|
|
|
if plugins :
|
|
|
|
self.init_plugins(plugins)
|
|
|
|
self.init_plugins(plugins)
|
|
|
|
# def setLogger(self,_logger):
|
|
|
|
|
|
|
|
# self._logger = _logger
|
|
|
|
|
|
|
|
# def log (self,**_args):
|
|
|
|
|
|
|
|
# if self._logger :
|
|
|
|
|
|
|
|
# _date = str(datetime.now())
|
|
|
|
|
|
|
|
# _data = dict({'pid':os.getpid(),'date':_date[:10],'time':_date[11:19]},**_args)
|
|
|
|
|
|
|
|
# for key in _data :
|
|
|
|
|
|
|
|
# if type(_data[key]) == list :
|
|
|
|
|
|
|
|
# _data[key] = [_item.__name__ if type(_item).__name__== 'function' else _item for _item in _data[key]]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# _data[key] = str(_data[key]) if type(_data[key]) not in [list,dict] else json.dumps(_data[key])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# self._logger.write(pd.DataFrame([_data])) #,table=self._logTable)
|
|
|
|
|
|
|
|
# def _init_plugins(self,_items):
|
|
|
|
|
|
|
|
# """
|
|
|
|
|
|
|
|
# This function will load pipelined functions as a plugin loader
|
|
|
|
|
|
|
|
# """
|
|
|
|
|
|
|
|
# registry.plugins.init()
|
|
|
|
|
|
|
|
# self._plugins = PluginLoader(registry=registry.plugins)
|
|
|
|
|
|
|
|
# [self._plugins.set(_name) for _name in _items]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# self.log(action='init-plugins',object=self.getClassName(self),input =[_name for _name in _items])
|
|
|
|
|
|
|
|
# # if 'path' in _args and 'names' in _args :
|
|
|
|
|
|
|
|
# # self._plugins = PluginLoader(**_args)
|
|
|
|
|
|
|
|
# # else:
|
|
|
|
|
|
|
|
# # self._plugins = PluginLoader(registry=registry.plugins)
|
|
|
|
|
|
|
|
# # [self._plugins.set(_pointer) for _pointer in _args]
|
|
|
|
|
|
|
|
# #
|
|
|
|
|
|
|
|
# # @TODO: We should have a way to log what plugins are loaded and ready to use
|
|
|
|
|
|
|
|
def meta (self,**_args):
|
|
|
|
def meta (self,**_args):
|
|
|
|
if hasattr(self._agent,'meta') :
|
|
|
|
if hasattr(self._agent,'meta') :
|
|
|
|
return self._agent.meta(**_args)
|
|
|
|
return self._agent.meta(**_args)
|
|
|
|
return []
|
|
|
|
return []
|
|
|
|
def getClassName (self,_object):
|
|
|
|
|
|
|
|
return '.'.join([_object.__class__.__module__,_object.__class__.__name__])
|
|
|
|
|
|
|
|
def close(self):
|
|
|
|
def close(self):
|
|
|
|
if hasattr(self._agent,'close') :
|
|
|
|
if hasattr(self._agent,'close') :
|
|
|
|
self._agent.close()
|
|
|
|
self._agent.close()
|
|
|
|
def apply(self):
|
|
|
|
# def apply(self):
|
|
|
|
"""
|
|
|
|
# """
|
|
|
|
applying pre/post conditions given a pipeline expression
|
|
|
|
# applying pre/post conditions given a pipeline expression
|
|
|
|
"""
|
|
|
|
# """
|
|
|
|
for _pointer in self._plugins :
|
|
|
|
# for _pointer in self._plugins :
|
|
|
|
_data = _pointer(_data)
|
|
|
|
# _data = _pointer(_data)
|
|
|
|
time.sleep(1)
|
|
|
|
|
|
|
|
def apply(self,_query):
|
|
|
|
def apply(self,_query):
|
|
|
|
if hasattr(self._agent,'apply') :
|
|
|
|
if hasattr(self._agent,'apply') :
|
|
|
|
return self._agent.apply(_query)
|
|
|
|
return self._agent.apply(_query)
|
|
|
|
@ -133,173 +62,70 @@ class IReader(IO):
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
def __init__(self,**_args):
|
|
|
|
def __init__(self,**_args):
|
|
|
|
super().__init__(**_args)
|
|
|
|
super().__init__(**_args)
|
|
|
|
self._args = _args['args']if 'args' in _args else None
|
|
|
|
|
|
|
|
def _stream (self,_data ):
|
|
|
|
|
|
|
|
# self.log(action='streaming',object=self._agent._engine.name, input= type(_data).__name__)
|
|
|
|
|
|
|
|
_shape = []
|
|
|
|
|
|
|
|
for _segment in _data :
|
|
|
|
|
|
|
|
_shape += list(_segment.shape)
|
|
|
|
|
|
|
|
if self._plugins :
|
|
|
|
|
|
|
|
# yield self._plugins.apply(_segment,self.log)
|
|
|
|
|
|
|
|
yield self._ixloader.visitor(_data,self.log)
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
|
|
|
yield _segment
|
|
|
|
|
|
|
|
_objectName = '.'.join([self._agent.__class__.__module__,self._agent.__class__.__name__])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_input = {'shape':_shape}
|
|
|
|
|
|
|
|
if hasattr(self._agent,'_table') :
|
|
|
|
|
|
|
|
_input['table'] = self._agent._table
|
|
|
|
|
|
|
|
self.log(action='streaming',object=_objectName, input= _input)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def read(self,**_args):
|
|
|
|
def read(self,**_args):
|
|
|
|
|
|
|
|
|
|
|
|
if 'plugins' in _args :
|
|
|
|
if 'plugins' in _args :
|
|
|
|
self.init_plugins(_args['plugins'])
|
|
|
|
self.init_plugins(_args['plugins'])
|
|
|
|
|
|
|
|
|
|
|
|
if self._args :
|
|
|
|
_data = self._agent.read(**_args)
|
|
|
|
_data = self._agent.read(**self._args)
|
|
|
|
# if self._plugins and self._plugins.ratio() > 0 :
|
|
|
|
else:
|
|
|
|
# _data = self._plugins.apply(_data)
|
|
|
|
_data = self._agent.read(**_args)
|
|
|
|
#
|
|
|
|
|
|
|
|
# output data
|
|
|
|
_objectName = '.'.join([self._agent.__class__.__module__,self._agent.__class__.__name__])
|
|
|
|
|
|
|
|
if types.GeneratorType == type(_data):
|
|
|
|
|
|
|
|
return self._stream(_data)
|
|
|
|
|
|
|
|
# if self._plugins :
|
|
|
|
|
|
|
|
# return self._stream(_data)
|
|
|
|
|
|
|
|
# else:
|
|
|
|
|
|
|
|
# _count = 0
|
|
|
|
|
|
|
|
# for _segment in _data :
|
|
|
|
|
|
|
|
# _count += 1
|
|
|
|
|
|
|
|
# yield _segment
|
|
|
|
|
|
|
|
# self.log(action='streaming',object=_objectName, input= {'segments':_count})
|
|
|
|
|
|
|
|
# return _data
|
|
|
|
|
|
|
|
elif type(_data) == pd.DataFrame :
|
|
|
|
|
|
|
|
_shape = _data.shape #[0,0] if not _data.shape[] else list(_data.shape)
|
|
|
|
|
|
|
|
_input = {'shape':_shape}
|
|
|
|
|
|
|
|
if hasattr(self._agent,'_table') :
|
|
|
|
|
|
|
|
_input['table'] = self._agent._table
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self.log(action='read',object=_objectName, input=_input)
|
|
|
|
|
|
|
|
_data = self._ixloader.visitor(_data)
|
|
|
|
|
|
|
|
return _data
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
# applying the the design pattern
|
|
|
|
|
|
|
|
_data = self._ixloader.visitor(_data)
|
|
|
|
|
|
|
|
return _data
|
|
|
|
class IWriter(IO):
|
|
|
|
class IWriter(IO):
|
|
|
|
lock = RLock()
|
|
|
|
def __init__(self,**_args): #_agent,pipeline=None):
|
|
|
|
def __init__(self,**_args):
|
|
|
|
super().__init__(**_args) #_agent,pipeline)
|
|
|
|
super().__init__(**_args)
|
|
|
|
|
|
|
|
def write(self,_data,**_args):
|
|
|
|
def write(self,_data,**_args):
|
|
|
|
|
|
|
|
# if 'plugins' in _args :
|
|
|
|
|
|
|
|
# self._init_plugins(_args['plugins'])
|
|
|
|
if 'plugins' in _args :
|
|
|
|
if 'plugins' in _args :
|
|
|
|
self._init_plugins(_args['plugins'])
|
|
|
|
self.init_plugins(_args['plugins'])
|
|
|
|
# if self._plugins and self._plugins.ratio() > 0 :
|
|
|
|
|
|
|
|
# _logs = []
|
|
|
|
|
|
|
|
# _data = self._plugins.apply(_data,_logs,self.log)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# [self.log(**_item) for _item in _logs]
|
|
|
|
self._ixloader.visitor(_data)
|
|
|
|
try:
|
|
|
|
self._agent.write(_data,**_args)
|
|
|
|
# IWriter.lock.acquire()
|
|
|
|
|
|
|
|
_data = self._ixloader.visitor(_data)
|
|
|
|
|
|
|
|
self._agent.write(_data,**_args)
|
|
|
|
|
|
|
|
finally:
|
|
|
|
|
|
|
|
# IWriter.lock.release()
|
|
|
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#
|
|
|
|
#
|
|
|
|
# The ETL object in its simplest form is an aggregation of read/write objects
|
|
|
|
# The ETL object in its simplest form is an aggregation of read/write objects
|
|
|
|
# @TODO: ETL can/should aggregate a writer as a plugin and apply it as a process
|
|
|
|
# @TODO: ETL can/should aggregate a writer as a plugin and apply it as a process
|
|
|
|
|
|
|
|
|
|
|
|
class IETL(BaseIO) :
|
|
|
|
class IETL(IReader) :
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
This class performs an ETL operation by ineriting a read and adding writes as pipeline functions
|
|
|
|
This class performs an ETL operation by ineriting a read and adding writes as pipeline functions
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
def __init__(self,**_args):
|
|
|
|
def __init__(self,**_args):
|
|
|
|
# _source = _args['source']
|
|
|
|
super().__init__(agent=transport.get.reader(**_args['source']),plugins=None)
|
|
|
|
# _plugins = _source['plugins'] if 'plugins' in _source else None
|
|
|
|
if 'target' in _args:
|
|
|
|
|
|
|
|
self._targets = _args['target'] if type(_args['target']) == list else [_args['target']]
|
|
|
|
# # super().__init__(transport.get.reader(**_args['source']))
|
|
|
|
|
|
|
|
# super().__init__(agent=transport.get.reader(**_source),plugins=_plugins)
|
|
|
|
|
|
|
|
# # _logger =
|
|
|
|
|
|
|
|
# if 'target' in _args:
|
|
|
|
|
|
|
|
# self._targets = _args['target'] if type(_args['target']) == list else [_args['target']]
|
|
|
|
|
|
|
|
# else:
|
|
|
|
|
|
|
|
# self._targets = []
|
|
|
|
|
|
|
|
# self.jobs = []
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# #
|
|
|
|
|
|
|
|
# # If the parent is already multiprocessing
|
|
|
|
|
|
|
|
# if 'token' in _source :
|
|
|
|
|
|
|
|
# self._logToken = _source['token']
|
|
|
|
|
|
|
|
# self._sourceArgs = _source['args'] if 'args' in _source else None
|
|
|
|
|
|
|
|
# self._hasParentProcess = False if 'hasParentProcess' not in _args else _args['hasParentProcess']
|
|
|
|
|
|
|
|
super().__init__()
|
|
|
|
|
|
|
|
self._source = _args['source']
|
|
|
|
|
|
|
|
self._targets= _args['target'] if _args['target'] == list else [_args['target']]
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
# ETL Initialization, we should provide some measure of context ...
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# def run(self) :
|
|
|
|
|
|
|
|
# """
|
|
|
|
|
|
|
|
# We should apply the etl here, if we are in multiprocessing mode
|
|
|
|
|
|
|
|
# """
|
|
|
|
|
|
|
|
# return self.read()
|
|
|
|
|
|
|
|
def run(self,**_args):
|
|
|
|
|
|
|
|
# _data = super().read(**_args) if not self._sourceArgs else super().read(**self._sourceArgs)
|
|
|
|
|
|
|
|
# self._targets = [transport.get.writer(**_kwargs) for _kwargs in self._targets]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_reader = transport.get.reader(**self._source)
|
|
|
|
|
|
|
|
if hasattr(_reader,'_logger') :
|
|
|
|
|
|
|
|
self.setLogger(_reader._logger)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self.log(action='init-etl',input={'source':self._source,'target':self._targets})
|
|
|
|
|
|
|
|
_data = _reader.read(**self._source['args'])if 'args' in self._source else _reader.read()
|
|
|
|
|
|
|
|
_reader.close()
|
|
|
|
|
|
|
|
_writers = [transport.get.writer(**_kwargs) for _kwargs in self._targets]
|
|
|
|
|
|
|
|
_schema = [] if not getattr(_reader._agent,'_table') else _reader.meta()
|
|
|
|
|
|
|
|
if types.GeneratorType == type(_data):
|
|
|
|
|
|
|
|
_index = 0
|
|
|
|
|
|
|
|
for _segment in _data :
|
|
|
|
|
|
|
|
_index += 1
|
|
|
|
|
|
|
|
for _writer in _writers :
|
|
|
|
|
|
|
|
self.post(_segment,writer=_writer,index=_index,schema=_schema)
|
|
|
|
|
|
|
|
time.sleep(1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
else:
|
|
|
|
else:
|
|
|
|
|
|
|
|
self._targets = []
|
|
|
|
|
|
|
|
self.jobs = []
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
# If the parent is already multiprocessing
|
|
|
|
|
|
|
|
self._hasParentProcess = False if 'hasParentProcess' not in _args else _args['hasParentProcess']
|
|
|
|
|
|
|
|
def read(self,**_args):
|
|
|
|
|
|
|
|
_data = super().read(**_args)
|
|
|
|
|
|
|
|
_schema = super().meta()
|
|
|
|
|
|
|
|
for _kwargs in self._targets :
|
|
|
|
|
|
|
|
if _schema :
|
|
|
|
|
|
|
|
_kwargs['schema'] = _schema
|
|
|
|
|
|
|
|
self.post(_data,**_kwargs)
|
|
|
|
|
|
|
|
|
|
|
|
for _writer in _writers :
|
|
|
|
|
|
|
|
self.post(_data,writer=_writer,schema=_schema)
|
|
|
|
|
|
|
|
# pass
|
|
|
|
|
|
|
|
return _data
|
|
|
|
return _data
|
|
|
|
# return _data
|
|
|
|
def run(self) :
|
|
|
|
|
|
|
|
return self.read()
|
|
|
|
def post (self,_data,**_args) :
|
|
|
|
def post (self,_data,**_args) :
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
This function returns an instance of a process that will perform the write operation
|
|
|
|
This function returns an instance of a process that will perform the write operation
|
|
|
|
:_args parameters associated with writer object
|
|
|
|
:_args parameters associated with writer object
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
#writer = transport.get.writer(**_args)
|
|
|
|
writer = transport.get.writer(**_args)
|
|
|
|
_input = {}
|
|
|
|
if 'schema' in _args :
|
|
|
|
try:
|
|
|
|
writer.write(_data,schema=_args['schema'])
|
|
|
|
_action = 'post'
|
|
|
|
else:
|
|
|
|
_shape = dict(zip(['rows','columns'],_data.shape))
|
|
|
|
writer.write(_data)
|
|
|
|
_index = _args['index'] if 'index' in _args else 0
|
|
|
|
writer.close()
|
|
|
|
|
|
|
|
|
|
|
|
writer = _args['writer']
|
|
|
|
|
|
|
|
_schema= _args['schema']
|
|
|
|
|
|
|
|
#
|
|
|
|
|
|
|
|
# -- things to log
|
|
|
|
|
|
|
|
_input = {'shape':_shape,'segment':_index}
|
|
|
|
|
|
|
|
if hasattr(writer._agent,'_table'):
|
|
|
|
|
|
|
|
_input['table'] = writer._agent._table
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
for _item in _schema :
|
|
|
|
|
|
|
|
if _item['type'] in ['INTEGER','BIGINT','INT'] :
|
|
|
|
|
|
|
|
_column = _item['name']
|
|
|
|
|
|
|
|
_data[_column] = _data[_column].copy().fillna(0).astype(np.int64)
|
|
|
|
|
|
|
|
writer.write(_data,schema=_schema)
|
|
|
|
|
|
|
|
except Exception as e:
|
|
|
|
|
|
|
|
_action = 'post-error'
|
|
|
|
|
|
|
|
_input['error'] = str(e)
|
|
|
|
|
|
|
|
print ([e])
|
|
|
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self.log(action=_action,object=writer._agent.__module__, input= _input)
|
|
|
|
|
|
|
|
|
|
|
|
|
Steve L. Nyemba
Steve Nyemba