[WIP] Hecuba and dislib integration

.travis.yml

@@ -5,7 +5,7 @@ sudo: required
 
 branches:
   only:
-    - master
+    - test_compss
     - /^release-.*/
 
 services:
@@ -14,24 +14,27 @@ env:
   global:
     - REGISTRY_USER=compss
     - secure: ""
+    - TEST_CASSANDRA_VERSION=3.11.4
 
 before_script:
-    - docker build --tag bscwdc/dislib .
-    - docker run $(bash <(curl -s https://codecov.io/env)) -d --name dislib bscwdc/dislib
-
-script: "docker exec dislib /dislib/run_ci_checks.sh"
-
-after_script:
-  - docker images
-  - docker exec dislib /dislib/bin/print_tests_logs.sh
-
-before_deploy:
-  - docker login -u "$REGISTRY_USER" -p "$REGISTRY_PASS"
-  - docker tag bscwdc/dislib bscwdc/dislib:latest
-deploy:
-  provider: script
-  script: docker push bscwdc/dislib:latest
-  on:
-    branch: master
+    - source launch_cassandra.sh
+    - docker build --tag emebemb/dislib_hecuba_compss_production:0.2 .
+    - docker run -it --network cassandra_bridge -d --name dislib emebemb/dislib_hecuba_compss_production:0.2
+
+
+script: "docker exec -e CONTACT_NAMES='cassandra_container' -e NODE_PORT=9042 dislib /dislib/run_tests.sh"
+
+#after_script:
+#  - docker images
+#  - docker exec dislib /dislib/bin/print_tests_logs.sh
+#
+#before_deploy:
+#  - docker login -u "$REGISTRY_USER" -p "$REGISTRY_PASS"
+#  - docker tag bscwdc/dislib bscwdc/dislib:latest
+#deploy:
+#  provider: script
+#  script: docker push bscwdc/dislib:latest
+#  on:
+#    branch: master
 
 

Dockerfile

@@ -1,6 +1,17 @@
-FROM bscwdc/dislib-base:latest
+#FROM bscwdc/dislib-base:latest
+FROM adrianespejo/dislib_hecuba:0.1
 MAINTAINER COMPSs Support <support-compss@bsc.es>
 
+#RUN apt-get update -y && apt-get update
+#RUN apt-get install -y cmake python3-dev libpython3-dev gcc-4.8 libtool python3-numpy python3-pip python3-setuptools
+#RUN curl -L https://github.com/bsc-dd/hecuba/archive/NumpyWritePartitions.tar.gz | tar -xz
+
+#WORKDIR hecuba-NumpyWritePartitions
+#RUN python3 -m pip install -r requirements.txt
+#RUN python3 setup.py install
+WORKDIR /
+
+#RUN rm -rf dislib/
 COPY . dislib/
 
 ENV PYTHONPATH=$PYTHONPATH:/dislib

counter

@@ -0,0 +1 @@
+2

dislib/__init__.py

@@ -1,7 +1,7 @@
 import os
 
 from dislib.data.array import random_array, apply_along_axis, array, zeros, \
-    full
+    full, load_from_hecuba
 from dislib.data.io import load_svmlight_file, load_npy_file, load_txt_file
 from dislib.math import kron
 
@@ -27,5 +27,5 @@
         __version__ = 'unknown'
 
 __all__ = ['load_txt_file', 'load_svmlight_file', 'random_array',
-           'apply_along_axis', 'array', 'load_npy_file', 'zeros', 'kron',
+           'apply_along_axis', 'array', 'load_from_hecuba', 'load_npy_file', 'zeros', 'kron',
            'full']

dislib/cluster/kmeans/base.py

@@ -1,6 +1,6 @@
 import numpy as np
 from pycompss.api.api import compss_wait_on
-from pycompss.api.parameter import COLLECTION_IN, Depth, Type
+from pycompss.api.parameter import INOUT, COLLECTION_IN, Depth, Type
 from pycompss.api.task import task
 from scipy.sparse import csr_matrix
 from sklearn.base import BaseEstimator
@@ -10,10 +10,8 @@
 
 from dislib.data.array import Array
 
-
 class KMeans(BaseEstimator):
     """ Perform K-means clustering.
-
     Parameters
     ----------
     n_clusters : int, optional (default=8)
@@ -22,7 +20,6 @@ class KMeans(BaseEstimator):
     init : {'random', nd-array or sparse matrix}, optional (default='random')
         Method of initialization, defaults to 'random', which generates
         random centers at the beginning.
-
         If an nd-array or sparse matrix is passed, it should be of shape
         (n_clusters, n_features) and gives the initial centers.
     max_iter : int, optional (default=10)
@@ -37,14 +34,12 @@ class KMeans(BaseEstimator):
         for centroid initialization.
     verbose: boolean, optional (default=False)
         Whether to print progress information.
-
     Attributes
     ----------
     centers : ndarray
         Computed centroids.
     n_iter : int
         Number of iterations performed.
-
     Examples
     --------
     >>> from dislib.cluster import KMeans
@@ -73,14 +68,12 @@ def __init__(self, n_clusters=8, init='random', max_iter=10, tol=1e-4,
 
     def fit(self, x, y=None):
         """ Compute K-means clustering.
-
         Parameters
         ----------
         x : ds-array
             Samples to cluster.
         y : ignored
             Not used, present here for API consistency by convention.
-
         Returns
         -------
         self : KMeans
@@ -95,6 +88,7 @@ def fit(self, x, y=None):
             old_centers = self.centers.copy()
             partials = []
 
+
             for row in x._iterator(axis=0):
                 partial = _partial_sum(row._blocks, old_centers)
                 partials.append(partial)
@@ -108,14 +102,12 @@ def fit(self, x, y=None):
 
     def fit_predict(self, x, y=None):
         """ Compute cluster centers and predict cluster index for each sample.
-
         Parameters
         ----------
         x : ds-array
             Samples to cluster.
         y : ignored
             Not used, present here for API consistency by convention.
-
         Returns
         -------
         labels : ds-array, shape=(n_samples, 1)
@@ -127,12 +119,10 @@ def fit_predict(self, x, y=None):
 
     def predict(self, x):
         """ Predict the closest cluster each sample in the data belongs to.
-
         Parameters
         ----------
         x : ds-array
             New data to predict.
-
         Returns
         -------
         labels : ds-array, shape=(n_samples, 1)
@@ -190,10 +180,10 @@ def _init_centers(self, n_features, sparse):
 
 
 @task(blocks={Type: COLLECTION_IN, Depth: 2}, returns=np.array)
+#@task(blocks=INOUT, returns=np.array)
 def _partial_sum(blocks, centers):
     partials = np.zeros((centers.shape[0], 2), dtype=object)
     arr = Array._merge_blocks(blocks)
-
     close_centers = pairwise_distances(arr, centers).argmin(axis=1)
 
     for center_idx, _ in enumerate(centers):
@@ -204,6 +194,8 @@ def _partial_sum(blocks, centers):
     return partials
 
 
+
+
 @task(returns=dict)
 def _merge(*data):
     accum = data[0].copy()
@@ -217,4 +209,4 @@ def _merge(*data):
 @task(blocks={Type: COLLECTION_IN, Depth: 2}, returns=np.array)
 def _predict(blocks, centers):
     arr = Array._merge_blocks(blocks)
-    return pairwise_distances(arr, centers).argmin(axis=1).reshape(-1, 1)
+    return pairwise_distances(arr, centers).argmin(axis=1).reshape(-1, 1)

dislib/data/__init__.py

@@ -1,6 +1,6 @@
 from dislib.data.array import array, random_array, apply_along_axis, zeros, \
-    full
+    full, load_from_hecuba
 from dislib.data.io import load_txt_file, load_npy_file, load_svmlight_file
 
 __all__ = ['load_txt_file', 'load_svmlight_file', 'array', 'random_array',
-           'apply_along_axis', 'load_npy_file', 'zeros', 'full']
+           'apply_along_axis', 'load_from_hecuba', 'load_npy_file', 'zeros', 'full']

dislib/data/array.py

@@ -1,7 +1,10 @@
+import itertools
+import uuid
 import operator
 from collections import defaultdict
 
 import numpy as np
+import importlib
 from pycompss.api.api import compss_wait_on, compss_delete_object
 from pycompss.api.parameter import Type, COLLECTION_IN, Depth, \
     COLLECTION_INOUT, INOUT
@@ -10,6 +13,12 @@
 from scipy.sparse import issparse, csr_matrix
 from sklearn.utils import check_random_state
 
+if importlib.util.find_spec("hecuba"):
+    try:
+        from hecuba.hnumpy import StorageNumpy
+    except Exception:
+        pass
+from pprint import pprint
 from math import ceil
 
 
@@ -109,6 +118,9 @@ def __matmul__(self, x):
                      reg_shape=reg_shape, shape=shape, sparse=self._sparse)
 
     def __getitem__(self, arg):
+        if getattr(self, "_base_array", None) is not None:
+            return array(x=list(self._base_array[arg]),
+                         block_size=self._reg_shape)
 
         # return a single row
         if isinstance(arg, int):
@@ -205,6 +217,19 @@ def _merge_blocks(blocks):
         a single ndarray / sparse matrix.
         """
         sparse = None
+
+        try:
+            if blocks[0][0].__class__.__name__=="StorageNumpy":
+                res=[]
+                for block in blocks:
+                    value=list(block)
+                    line=np.concatenate(value,axis=1)
+                    res.append(line)
+                return np.concatenate(res)
+        except:
+            print("Block size no compatible with np.array.shape")
+
+
         b0 = blocks[0][0]
         if sparse is None:
             sparse = issparse(b0)
@@ -919,6 +944,39 @@ def collect(self, squeeze=True):
             res = np.squeeze(res)
         return res
 
+    def make_persistent(self, name):
+        """
+        Stores data in Hecuba.
+
+        Parameters
+        ----------
+        name : str
+            Name of the data.
+
+        Returns
+        -------
+        dsarray : ds-array
+            A distributed and persistent representation of the data
+            divided in blocks.
+        """
+        if self._sparse:
+            raise Exception("Data must not be a sparse matrix.")
+
+        x = self.collect()
+        persistent_data = StorageNumpy(input_array=x, name=name)
+        # self._base_array is used for much more efficient slicing.
+        # It does not take up more space since it is a reference to the db.
+        self._base_array = persistent_data
+
+        blocks = []
+        for block in self._blocks:
+            persistent_block = StorageNumpy(input_array=block, name=name,
+                                            storage_id=uuid.uuid4())
+            blocks.append(persistent_block)
+        self._blocks = blocks
+
+        return self
+
 
 def array(x, block_size):
     """
@@ -936,6 +994,8 @@ def array(x, block_size):
     dsarray : ds-array
         A distributed representation of the data divided in blocks.
     """
+    bn, bm = block_size
+
     sparse = issparse(x)
 
     if sparse:
@@ -958,8 +1018,6 @@ def array(x, block_size):
     if x.shape[0] < block_size[0] or x.shape[1] < block_size[1]:
         raise ValueError("Block size is greater than the array")
 
-    bn, bm = block_size
-
     blocks = []
     for i in range(0, x.shape[0], bn):
         row = [x[i: i + bn, j: j + bm] for j in range(0, x.shape[1], bm)]
@@ -972,6 +1030,38 @@ def array(x, block_size):
     return arr
 
 
+def load_from_hecuba(name, block_size):
+    """
+    Loads data from Hecuba.
+
+    Parameters
+    ----------
+    name : str
+        Name of the data.
+    block_size : (int, int)
+        Block sizes in number of samples.
+
+    Returns
+    -------
+    storagenumpy : StorageNumpy
+        A distributed and persistent representation of the data
+        divided in blocks.
+    """
+    persistent_data = StorageNumpy(name=name)
+
+    bn, bm = block_size
+
+    blocks = []
+    for block in persistent_data.np_split(block_size=(bn, bm)):
+        blocks.append(block)
+
+    arr = Array(blocks=blocks, top_left_shape=block_size,
+                reg_shape=block_size, shape=persistent_data.shape,
+                sparse=False)
+    arr._base_array = persistent_data
+    return arr
+
+
 def random_array(shape, block_size, random_state=None):
     """
     Returns a distributed array of random floats in the open interval [0.0,

killcompss.py

@@ -0,0 +1,22 @@
+#!/usr/bin/python
+import os
+import shutil
+import subprocess
+
+def main():
+    p = subprocess.Popen(['ps', '-ef'], stdout=subprocess.PIPE)
+    killed_count = -1
+    for line in p.stdout.readlines():
+        if 'compss' in line.decode() or 'COMPSs' in line.decode():
+            candidates = line.decode().split(" ")[1:]
+            for cand in candidates:
+                if cand:
+                    pid = cand
+                    break
+            subprocess.Popen(['kill', '-9', pid])
+            killed_count += 1
+    print('%d total processes killed'%killed_count)
+
+
+if __name__ == "__main__":
+    main()