Reorganize repository & prepare pip package

Author: nalepae

.gitignore

@@ -1,4 +1,7 @@
 *.pkl
 .vscode
 __pycache__
-*checkpoints
+*checkpoints
+build
+dist
+*.egg-info

examples.ipynb docs/examples.ipynbexamples.ipynb renamed to docs/examples.ipynb

@@ -0,0 +1 @@
+__version__ = '0.1.0'

pandarallel/__init__.py

@@ -1,17 +1,19 @@
-import pandas as pd
-import pyarrow.plasma as plasma
-from pyarrow.lib import PlasmaStoreFull
-import multiprocessing
-import itertools
-from concurrent.futures import ProcessPoolExecutor
+import pandas as _pd
+import pyarrow.plasma as _plasma
+from pyarrow.lib import PlasmaStoreFull as _PlasmaStoreFull
+import multiprocessing as _multiprocessing
+import itertools as _itertools
+from concurrent.futures import ProcessPoolExecutor as _ProcessPoolExecutor
 
-plasma_store_ctx = None
-plasma_store_name = None
+__all__ = ['shm_size', 'nb_workers']
+
+_plasma_store_ctx = None
+_plasma_store_name = None
 
 shm_size = int(2e9) # 2 Go
-nb_workers = multiprocessing.cpu_count()
+nb_workers = _multiprocessing.cpu_count()
 
-def chunk(nb_elem, nb_chunks):
+def _chunk(nb_elem, nb_chunks):
     quotient = nb_elem // nb_chunks
     remainder = nb_elem % nb_chunks
 
@@ -22,7 +24,7 @@ def chunk(nb_elem, nb_chunks):
                             quotient + remainder for quotient, remainder
                             in zip(quotients, remainders)
                         ]
-    accumulated = list(itertools.accumulate(nb_elems_per_chunk))
+    accumulated = list(_itertools.accumulate(nb_elems_per_chunk))
     shifted_accumulated = accumulated.copy()
     shifted_accumulated.insert(0, 0)
     shifted_accumulated.pop()
@@ -32,25 +34,25 @@ def chunk(nb_elem, nb_chunks):
             in zip(shifted_accumulated, accumulated)
         ]
 
-def parallel(func):
+def _parallel(func):
     def wrapper(*args, **kwargs):
         try:
-            global plasma_store_ctx
-            global plasma_store_name
+            global _plasma_store_ctx
+            global _plasma_store_name
 
-            if not plasma_store_ctx:
+            if not _plasma_store_ctx:
                 mem_mo = round(shm_size / 1e6, 2)
                 msg = f"New pandarallel shared memory created - \
 Size: {mem_mo} Mo"
                 print(msg)
-                plasma_store_ctx = plasma.start_plasma_store(shm_size)
-                plasma_store_name, _ = plasma_store_ctx.__enter__()
+                _plasma_store_ctx = _plasma.start_plasma_store(shm_size)
+                _plasma_store_name, _ = _plasma_store_ctx.__enter__()
 
             print(f"Running task on {nb_workers} workers")
 
             return func(*args, **kwargs)
 
-        except PlasmaStoreFull:
+        except _PlasmaStoreFull:
             msg = f"The pandarallel shared memory: \
 {round(shm_size / 1e6, 2)} Mo is too small to allow parallel computation. \
 Just after pandarallel import, please write: \
@@ -61,28 +63,28 @@ def wrapper(*args, **kwargs):
 
     return wrapper
 
-class Series:
+class _Series:
     @staticmethod
     def worker(plasma_store_name, object_id, chunk, func):
-        client = plasma.connect(plasma_store_name)
+        client = _plasma.connect(plasma_store_name)
         series = client.get(object_id)
         return client.put(series[chunk].map(func))
 
     @staticmethod
-    @parallel
+    @_parallel
     def map(data, func):
-        client = plasma.connect(plasma_store_name)
-        chunks = chunk(data.size, nb_workers)
+        client = _plasma.connect(_plasma_store_name)
+        chunks = _chunk(data.size, nb_workers)
         object_id = client.put(data)
 
-        with ProcessPoolExecutor(max_workers=nb_workers) as executor:
+        with _ProcessPoolExecutor(max_workers=nb_workers) as executor:
             futures = [
-                        executor.submit(Series.worker, plasma_store_name,
-                                        object_id, chunk, func)
-                        for chunk in chunks
+                        executor.submit(_Series.worker, _plasma_store_name,
+                                        object_id, _chunk, func)
+                        for _chunk in chunks
                     ]
 
-        result = pd.concat([
+        result = _pd.concat([
                             client.get(future.result())
                             for future in futures
                         ], copy=False)
@@ -91,51 +93,51 @@ def map(data, func):
 
         return result
 
-pd.Series.parallel_map = Series.map
+_pd.Series.parallel_map = _Series.map
 
-class DataFrameGroupBy:
+class _DataFrameGroupBy:
     @staticmethod
     def worker(plasma_store_name, object_id, func):
-        client = plasma.connect(plasma_store_name)
+        client = _plasma.connect(plasma_store_name)
         df = client.get(object_id)
         return client.put(func(df))
 
     @staticmethod
-    @parallel
+    @_parallel
     def apply(data, func):
-        client = plasma.connect(plasma_store_name)
+        client = _plasma.connect(_plasma_store_name)
         keys = data.groups.keys()
 
-        with ProcessPoolExecutor(max_workers=nb_workers) as executor:
+        with _ProcessPoolExecutor(max_workers=nb_workers) as executor:
             futures = [
-                        executor.submit(DataFrameGroupBy.worker,
-                                        plasma_store_name,
+                        executor.submit(_DataFrameGroupBy.worker,
+                                        _plasma_store_name,
                                         client.put(data.get_group(key)),
                                         func)
                         for key in keys
                     ]
 
-        result = pd.DataFrame([
+        result = _pd.DataFrame([
                                 client.get(future.result())
                                 for future in futures
-                            ], index=pd.Series(list(data.grouper),
+                            ], index=_pd.Series(list(data.grouper),
                                name=data.keys))
 
         client.delete(client.list().keys())
 
         return result
 
-pd.core.groupby.DataFrameGroupBy.parallel_apply = DataFrameGroupBy.apply
+_pd.core.groupby.DataFrameGroupBy.parallel_apply = _DataFrameGroupBy.apply
 
-class DataFrame:
+class _DataFrame:
     @staticmethod
     def worker(plasma_store_name, object_id, chunk, func, **kwargs):
-        client = plasma.connect(plasma_store_name)
+        client = _plasma.connect(plasma_store_name)
         df = client.get(object_id)
         return client.put(df[chunk].apply(func, **kwargs))
 
     @staticmethod
-    @parallel
+    @_parallel
     def apply(data, func, **kwargs):
         axis = kwargs.get("axis", 0)
         if axis == 0:
@@ -144,18 +146,18 @@ def apply(data, func, **kwargs):
 Implementation of dataframe.parallel_apply with axis=0 will come soon."
             raise NotImplementedError(msg)
 
-        client = plasma.connect(plasma_store_name)
-        chunks = chunk(data.shape[0], nb_workers)
+        client = _plasma.connect(_plasma_store_name)
+        chunks = _chunk(data.shape[0], nb_workers)
         object_id = client.put(data)
 
-        with ProcessPoolExecutor(max_workers=nb_workers) as executor:
+        with _ProcessPoolExecutor(max_workers=nb_workers) as executor:
             futures = [
-                        executor.submit(DataFrame.worker, plasma_store_name,
-                                        object_id, chunk, func, **kwargs)
-                        for chunk in chunks
+                        executor.submit(_DataFrame.worker, _plasma_store_name,
+                                        object_id, _chunk, func, **kwargs)
+                        for _chunk in chunks
                     ]
 
-        result = pd.concat([
+        result = _pd.concat([
                             client.get(future.result())
                             for future in futures
                         ], copy=False)
@@ -165,4 +167,4 @@ def apply(data, func, **kwargs):
         return result        
 
 
-pd.DataFrame.parallel_apply = DataFrame.apply
+_pd.DataFrame.parallel_apply = _DataFrame.apply

pandarallel.py pandarallel/pandarallel.pypandarallel.py renamed to pandarallel/pandarallel.py

@@ -0,0 +1,14 @@
+from setuptools import setup, find_packages
+
+import pandarallel
+
+setup(
+    name='pandarallel',
+    version=pandarallel.__version__,
+    packages=find_packages(),
+    author='Manu NALEPA',
+    author_email='nalepae@gmail.com',
+    description='An easy to use library to speed up computation (by parallelizing on multi CPUs) with pandas.',
+    long_description=open('README.md').read(),
+    url='https://github.com/nalepae/pandarallel',
+)

setup.py

@@ -1,6 +1,6 @@
-import pandarallel
+import pandarallel.pandarallel
 
-import pandas as pd
+import pandas as _pd
 import numpy as np
 import math
 
@@ -14,12 +14,12 @@ def func_for_dataframe_groupby_apply(df):
     dum = 0
     for item in df.b:
         dum += math.log10(math.sqrt(math.exp(item**2)))
-        
+
     return dum / len(df.b)
 
 def test_dataframe_apply():
     df_size = int(1e1)
-    df = pd.DataFrame(dict(a=np.random.randint(1, 8, df_size),
+    df = _pd.DataFrame(dict(a=np.random.randint(1, 8, df_size),
                         b=np.random.rand(df_size)))
 
     res = df.apply(func_for_dataframe_apply, axis=1)
@@ -28,15 +28,15 @@ def test_dataframe_apply():
 
 def test_series_map():
     df_size = int(1e1)
-    df = pd.DataFrame(dict(a=np.random.rand(df_size) + 1))
+    df = _pd.DataFrame(dict(a=np.random.rand(df_size) + 1))
 
     res = df.a.map(func_for_series_map)
     res_parallel = df.a.parallel_map(func_for_series_map)
     assert res.equals(res_parallel)
 
 def test_dataframe_groupby_apply():
     df_size = int(1e1)
-    df = pd.DataFrame(dict(a=np.random.randint(1, 8, df_size),
+    df = _pd.DataFrame(dict(a=np.random.randint(1, 8, df_size),
                            b=np.random.rand(df_size)))
 
     res = df.groupby("a").apply(func_for_dataframe_groupby_apply)