bin/transport

@@ -24,8 +24,7 @@ from multiprocessing import Process
 
 import os
 import transport
-# from transport import etl
-from transport.iowrapper import IETL
+from transport import etl
 # from transport import providers
 import typer
 from typing_extensions import Annotated
@@ -61,13 +60,10 @@ def apply (path:Annotated[str,typer.Argument(help="path of the configuration fil
             _config = [_config[ int(index)]]
         jobs = []            
         for _args in _config :
-            # pthread = etl.instance(**_args) #-- automatically starts the process
-            _worker = IETL(**_args)
-            pthread = Process(target=_worker.run)
-            pthread.start()
+            pthread = etl.instance(**_args) #-- automatically starts the process
             jobs.append(pthread)
         #
-        # @TODO: Log the number of processes started and estfrom transport impfrom transport impimated time
+        # @TODO: Log the number of processes started and estimated time
         while jobs :
              jobs = [pthread for pthread in jobs if pthread.is_alive()]
              time.sleep(1)
@@ -92,7 +88,6 @@ def version():
     """
 
     print (transport.__app_name__,'version ',transport.__version__)
-    print ()
     print (transport.__license__)
 
 @app.command()

info/__init__.py

@@ -1,6 +1,6 @@
 __app_name__  = 'data-transport'
 __author__ = 'The Phi Technology'
-__version__= '2.4.0'
+__version__= '2.2.6'
 __email__  = "info@the-phi.com"
 __license__=f"""
 Copyright 2010 - 2024, Steve L. Nyemba
@@ -15,9 +15,7 @@ THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR I
 
 __whatsnew__=f"""version {__version__}, focuses on collaborative environments like jupyter-base servers (apache zeppelin; jupyter notebook, jupyterlab, jupyterhub)
 
-    1. support for apache iceberg data warehouse using spark
-    2. Improved ETL & performance
-    3. bug fixes: mongodb
-
-    
+    1. simpler syntax to create readers/writers
+    2. auth-file registry that can be referenced using a label
+    3. duckdb support
 """

transport/__init__.py

@@ -18,7 +18,7 @@ Source Code is available under MIT License:
 """
 import numpy as np
 
-from transport import sql, nosql, cloud, other, warehouse
+from transport import sql, nosql, cloud, other
 import pandas as pd
 import json
 import os
@@ -33,7 +33,7 @@ PROVIDERS = {}
 
 def init():
     global PROVIDERS
-    for _module in [cloud,sql,nosql,other,warehouse] :
+    for _module in [cloud,sql,nosql,other] :
         for _provider_name in dir(_module) :
             if _provider_name.startswith('__') or _provider_name == 'common':
                 continue

transport/iowrapper.py

@@ -103,14 +103,6 @@ class IETL(IReader) :
         #
         # If the parent is already multiprocessing
         self._hasParentProcess = False if 'hasParentProcess' not in _args else _args['hasParentProcess']
-    def run(self) :
-        """
-        We should apply the etl here, if we are in multiprocessing mode
-        """
-        _data = super().read()
-        for _kwargs in self._targets :
-            self.post(_data,**_kwargs)
-
     def read(self,**_args):
         _data = super().read(**_args)
 

transport/providers/__init__.py

@@ -44,8 +44,7 @@ PGSQL	= POSTGRESQL
 
 AWS_S3  = 's3'
 RABBIT = RABBITMQ
-ICEBERG='iceberg'
-APACHE_ICEBERG = 'iceberg'
+
 
 # QLISTENER = 'qlistener'
     

transport/warehouse/__init__.py

@@ -1,7 +0,0 @@
-"""
-This namespace/package is intended to handle read/writes against data warehouse solutions like :
-    - apache iceberg
-    - clickhouse (...)
-"""
-
-from . import iceberg

transport/warehouse/iceberg.py

@@ -1,103 +0,0 @@
-from pyspark.sql import SparkSession
-import copy
-
-class Iceberg :
-    def __init__(self,**_args):
-        """
-        providing catalog meta information (you must get this from apache iceberg)
-        """
-        #
-        # @TODO:
-        #   Make arrangements for additional configuration elements 
-        #
-        self._session = SparkSession.builder.getOrCreate()
-        self._catalog = self._session.catalog
-        self._table = _args['table'] if 'table' in _args else None
-        
-        if 'catalog' in _args :
-            #
-            # Let us set the default catalog
-            self._catalog.setCurrentCatalog(_args['catalog'])
-            
-        else:
-            # No current catalog has been set ...
-            pass
-        if 'database' in _args :
-            self._database = _args['database']
-            self._catalog.setCurrentDatabase(self._database)
-        else:
-            #
-            # Should we set the default as the first one if available ?
-            #
-            pass
-        self._catalogName = self._catalog.currentCatalog()
-        self._databaseName = self._catalog.currentDatabase()
-    def meta (self,**_args) :
-        """
-        This function should return the schema of a table (only)
-        """
-        _schema = []
-        try:
-            _tableName = self._getPrefix(**_args) + f".{_args['table']}"
-            print (_tableName)
-            _tmp = self._session.table(_tableName).schema
-            _schema = _tmp.jsonValue()['fields']
-            for _item in _schema :
-                del _item['nullable'],_item['metadata']
-        except Exception as e:
-            
-            pass
-        return _schema
-    def _getPrefix (self,**_args):        
-        _catName = self._catalogName if 'catalog' not in _args else _args['catalog']
-        _datName = self._databaseName if 'database' not in _args else _args['database']
-        
-        return '.'.join([_catName,_datName])
-            
-    def has (self,**_args):
-        try:
-            _prefix = self._getPrefix(**_args)
-            if _prefix.endswith('.') :
-                return False
-            return _args['table'] in [_item.name for _item in self._catalog.listTables(_prefix)]
-        except Exception as e:
-            print (e)
-            return False
-    def apply(self,sql):
-        pass
-class Reader(Iceberg) :
-    def __init__(self,**_args):
-        super().__init__(**_args)
-    def read(self,**_args):
-        _table = self._table
-        _prefix = self._getPrefix(**_args)        
-        if 'table' in _args or _table:
-            _table = _args['table'] if 'table' in _args else _table
-            _table = _prefix + f'.{_table}'
-            return self._session.table(_table).toPandas()
-        else:
-            sql = _args['sql']
-            return self._session.sql(sql).toPandas()
-        pass
-class Writer (Iceberg):
-    """
-    Writing data to an Apache Iceberg data warehouse (using pyspark)
-    """
-    def __init__(self,**_args):
-        super().__init__(**_args)
-        self._mode = 'append' if 'mode' not in _args else _args['mode']
-        self._table = None if 'table' not in _args else _args['table']
-    def write(self,_data,**_args):
-        _prefix = self._getPrefix(**_args)
-        if 'table' not in _args and not self._table :
-            raise Exception (f"Table Name should be specified for catalog/database {_prefix}")
-        rdd = self._session.createDataFrame(_data)
-        _mode = self._mode if 'mode' not in _args else _args['mode']
-        _table = self._table if 'table' not in _args else _args['table']
-        
-        if not self.has(table=_table) :
-            _mode = 'overwrite'
-            rdd.write.format('iceberg').mode(_mode).saveAsTable(_table)
-        else:
-            _table = f'{_prefix}.{_table}'
-            rdd.write.format('iceberg').mode(_mode).save(_table)