You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by ni...@apache.org on 2018/08/28 00:35:15 UTC
[1/3] systemml git commit: [MINOR] Refactored the Python code to
match PEP8 guidelines and bugfix in Keras2DML for Python 3.5
Repository: systemml
Updated Branches:
refs/heads/master ddc3672f1 -> 9e7ee19a4
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/mllearn/keras2caffe.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/mllearn/keras2caffe.py b/src/main/python/systemml/mllearn/keras2caffe.py
index 3f4bddf..6e1e9c3 100755
--- a/src/main/python/systemml/mllearn/keras2caffe.py
+++ b/src/main/python/systemml/mllearn/keras2caffe.py
@@ -23,7 +23,8 @@
import numpy as np
-import os, math
+import os
+import math
from itertools import chain, imap
from ..converters import *
from ..classloader import *
@@ -36,7 +37,8 @@ try:
from py4j.java_gateway import JavaObject
from pyspark import SparkContext
except ImportError:
- raise ImportError('Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
+ raise ImportError(
+ 'Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
# --------------------------------------------------------------------------------------
# Design Document:
@@ -44,14 +46,17 @@ except ImportError:
#
# Part 1: Keras network to Caffe network conversion:
# - Core logic: model.layers.flatMap(layer => _parseJSONObject(_parseKerasLayer(layer)))
-# That is, for each layer, we first convert it into JSON format and then convert the JSON object into String
+# That is, for each layer, we first convert it into JSON format and then convert the JSON object into String
# - This is true for all the layers except the "specialLayers" (given in below hashmap). These are redirected to their custom parse function in _parseKerasLayer.
# - To add an activation, simply add the keras type to caffe type in supportedCaffeActivations.
# - To add a layer, add the corresponding caffe layer type in supportedLayers. If the layer accepts parameters then update layerParamMapping too.
# - The above logic is implemented in the function converKerasToCaffeNetwork
# --------------------------------------------------------------------------------------
-supportedCaffeActivations = {'relu':'ReLU', 'softmax':'Softmax', 'sigmoid':'Sigmoid' }
+supportedCaffeActivations = {
+ 'relu': 'ReLU',
+ 'softmax': 'Softmax',
+ 'sigmoid': 'Sigmoid'}
supportedLayers = {
keras.layers.InputLayer: 'Data',
keras.layers.Dense: 'InnerProduct',
@@ -63,22 +68,26 @@ supportedLayers = {
keras.layers.UpSampling2D: 'Upsample',
keras.layers.MaxPooling2D: 'Pooling',
keras.layers.AveragePooling2D: 'Pooling',
- keras.layers.SimpleRNN: 'RNN',
+ keras.layers.SimpleRNN: 'RNN',
keras.layers.LSTM: 'LSTM',
- keras.layers.Flatten: 'None',
+ keras.layers.Flatten: 'None',
keras.layers.BatchNormalization: 'None',
keras.layers.Activation: 'None'
- }
+}
+
def _getInboundLayers(layer):
in_names = []
# get inbound nodes to current layer (support newer as well as older APIs)
- inbound_nodes = layer.inbound_nodes if hasattr(layer, 'inbound_nodes') else layer._inbound_nodes
+ inbound_nodes = layer.inbound_nodes if hasattr(
+ layer, 'inbound_nodes') else layer._inbound_nodes
for node in inbound_nodes:
node_list = node.inbound_layers # get layers pointing to this node
in_names = in_names + node_list
# For Caffe2DML to reroute any use of Flatten layers
- return list(chain.from_iterable( [ _getInboundLayers(l) if isinstance(l, keras.layers.Flatten) else [ l ] for l in in_names ] ))
+ return list(chain.from_iterable([_getInboundLayers(l) if isinstance(
+ l, keras.layers.Flatten) else [l] for l in in_names]))
+
def _getCompensatedAxis(layer):
compensated_axis = layer.axis
@@ -91,265 +100,401 @@ def _getCompensatedAxis(layer):
compensated_axis = 1
return compensated_axis
-str_keys = [ 'name', 'type', 'top', 'bottom' ]
+
+str_keys = ['name', 'type', 'top', 'bottom']
+
+
def toKV(key, value):
- return str(key) + ': "' + str(value) + '"' if key in str_keys else str(key) + ': ' + str(value)
-
+ return str(key) + ': "' + str(value) + \
+ '"' if key in str_keys else str(key) + ': ' + str(value)
+
def _parseJSONObject(obj):
- rootName = obj.keys()[0]
- ret = ['\n', rootName, ' {']
- for key in obj[rootName]:
- if isinstance(obj[rootName][key], dict):
- ret = ret + [ '\n\t', key, ' {' ]
- for key1 in obj[rootName][key]:
- ret = ret + [ '\n\t\t', toKV(key1, obj[rootName][key][key1]) ]
- ret = ret + [ '\n\t', '}' ]
- elif isinstance(obj[rootName][key], list):
- for v in obj[rootName][key]:
- ret = ret + ['\n\t', toKV(key, v) ]
- else:
- ret = ret + ['\n\t', toKV(key, obj[rootName][key]) ]
- return ret + ['\n}' ]
-
+ rootName = obj.keys()[0]
+ ret = ['\n', rootName, ' {']
+ for key in obj[rootName]:
+ if isinstance(obj[rootName][key], dict):
+ ret = ret + ['\n\t', key, ' {']
+ for key1 in obj[rootName][key]:
+ ret = ret + ['\n\t\t', toKV(key1, obj[rootName][key][key1])]
+ ret = ret + ['\n\t', '}']
+ elif isinstance(obj[rootName][key], list):
+ for v in obj[rootName][key]:
+ ret = ret + ['\n\t', toKV(key, v)]
+ else:
+ ret = ret + ['\n\t', toKV(key, obj[rootName][key])]
+ return ret + ['\n}']
+
def _getBottomLayers(layer):
- return [ bottomLayer.name for bottomLayer in _getInboundLayers(layer) ]
+ return [bottomLayer.name for bottomLayer in _getInboundLayers(layer)]
def _parseActivation(layer, customLayerName=None):
- kerasActivation = keras.activations.serialize(layer.activation)
- if kerasActivation not in supportedCaffeActivations:
- raise TypeError('Unsupported activation ' + kerasActivation + ' for the layer:' + layer.name)
- if customLayerName is not None:
- return { 'layer':{'name':customLayerName, 'type':supportedCaffeActivations[kerasActivation], 'top':layer.name, 'bottom':layer.name }}
- else:
- return { 'layer':{'name':layer.name, 'type':supportedCaffeActivations[kerasActivation], 'top':layer.name, 'bottom':_getBottomLayers(layer) }}
+ kerasActivation = keras.activations.serialize(layer.activation)
+ if kerasActivation not in supportedCaffeActivations:
+ raise TypeError(
+ 'Unsupported activation ' +
+ kerasActivation +
+ ' for the layer:' +
+ layer.name)
+ if customLayerName is not None:
+ return {'layer': {'name': customLayerName,
+ 'type': supportedCaffeActivations[kerasActivation], 'top': layer.name, 'bottom': layer.name}}
+ else:
+ return {'layer': {'name': layer.name,
+ 'type': supportedCaffeActivations[kerasActivation], 'top': layer.name, 'bottom': _getBottomLayers(layer)}}
def _shouldParseActivation(layer):
- ignore_activation = [ keras.layers.SimpleRNN , keras.layers.LSTM ]
- return hasattr(layer, 'activation') and (type(layer) not in ignore_activation) and keras.activations.serialize(layer.activation) != 'linear'
+ ignore_activation = [keras.layers.SimpleRNN, keras.layers.LSTM]
+ return hasattr(layer, 'activation') and (type(
+ layer) not in ignore_activation) and keras.activations.serialize(layer.activation) != 'linear'
+
def _parseKerasLayer(layer):
- layerType = type(layer)
- if layerType in specialLayers:
- return specialLayers[layerType](layer)
- elif layerType == keras.layers.Activation:
- return [ _parseActivation(layer) ]
- param = layerParamMapping[layerType](layer)
- paramName = param.keys()[0]
- if layerType == keras.layers.InputLayer:
- ret = { 'layer': { 'name':layer.name, 'type':'Data', paramName:param[paramName], 'top':layer.name, 'top':'label' } }
- else:
- ret = { 'layer': { 'name':layer.name, 'type':supportedLayers[layerType], 'bottom':_getBottomLayers(layer), 'top':layer.name, paramName:param[paramName] } }
- return [ ret, _parseActivation(layer, layer.name + '_activation') ] if _shouldParseActivation(layer) else [ ret ]
+ layerType = type(layer)
+ if layerType in specialLayers:
+ return specialLayers[layerType](layer)
+ elif layerType == keras.layers.Activation:
+ return [_parseActivation(layer)]
+ param = layerParamMapping[layerType](layer)
+ paramName = param.keys()[0]
+ if layerType == keras.layers.InputLayer:
+ ret = {
+ 'layer': {
+ 'name': layer.name,
+ 'type': 'Data',
+ paramName: param[paramName],
+ 'top': layer.name,
+ 'top': 'label'}}
+ else:
+ ret = {
+ 'layer': {
+ 'name': layer.name,
+ 'type': supportedLayers[layerType],
+ 'bottom': _getBottomLayers(layer),
+ 'top': layer.name,
+ paramName: param[paramName]}}
+ return [ret, _parseActivation(
+ layer, layer.name + '_activation')] if _shouldParseActivation(layer) else [ret]
def _parseBatchNorm(layer):
- bnName = layer.name + '_1'
- config = layer.get_config()
- bias_term = 'true' if config['center'] else 'false'
- return [ { 'layer': { 'name':bnName, 'type':'BatchNorm', 'bottom':_getBottomLayers(layer), 'top':bnName, 'batch_norm_param':{'moving_average_fraction':layer.momentum, 'eps':layer.epsilon} } }, { 'layer': { 'name':layer.name, 'type':'Scale', 'bottom':bnName, 'top':layer.name, 'scale_param':{'bias_term':bias_term} } } ]
+ bnName = layer.name + '_1'
+ config = layer.get_config()
+ bias_term = 'true' if config['center'] else 'false'
+ return [{'layer': {'name': bnName, 'type': 'BatchNorm', 'bottom': _getBottomLayers(layer), 'top': bnName, 'batch_norm_param': {'moving_average_fraction': layer.momentum, 'eps': layer.epsilon}}}, {
+ 'layer': {'name': layer.name, 'type': 'Scale', 'bottom': bnName, 'top': layer.name, 'scale_param': {'bias_term': bias_term}}}]
+
# The special are redirected to their custom parse function in _parseKerasLayer
specialLayers = {
keras.layers.Flatten: lambda x: [],
keras.layers.BatchNormalization: _parseBatchNorm
- }
-
+}
+
+
def getConvParam(layer):
- stride = (1, 1) if layer.strides is None else layer.strides
- padding = [layer.kernel_size[0] / 2, layer.kernel_size[1] / 2] if layer.padding == 'same' else [0, 0]
- config = layer.get_config()
- return {'num_output':layer.filters,'bias_term':str(config['use_bias']).lower(),'kernel_h':layer.kernel_size[0], 'kernel_w':layer.kernel_size[1], 'stride_h':stride[0],'stride_w':stride[1],'pad_h':padding[0], 'pad_w':padding[1]}
+ stride = (1, 1) if layer.strides is None else layer.strides
+ padding = [
+ layer.kernel_size[0] /
+ 2,
+ layer.kernel_size[1] /
+ 2] if layer.padding == 'same' else [
+ 0,
+ 0]
+ config = layer.get_config()
+ return {'num_output': layer.filters, 'bias_term': str(config['use_bias']).lower(
+ ), 'kernel_h': layer.kernel_size[0], 'kernel_w': layer.kernel_size[1], 'stride_h': stride[0], 'stride_w': stride[1], 'pad_h': padding[0], 'pad_w': padding[1]}
+
def getUpSamplingParam(layer):
- return { 'size_h':layer.size[0], 'size_w':layer.size[1] }
+ return {'size_h': layer.size[0], 'size_w': layer.size[1]}
+
def getPoolingParam(layer, pool='MAX'):
- stride = (1, 1) if layer.strides is None else layer.strides
- padding = [layer.pool_size[0] / 2, layer.pool_size[1] / 2] if layer.padding == 'same' else [0, 0]
- return {'pool':pool, 'kernel_h':layer.pool_size[0], 'kernel_w':layer.pool_size[1], 'stride_h':stride[0],'stride_w':stride[1],'pad_h':padding[0], 'pad_w':padding[1]}
+ stride = (1, 1) if layer.strides is None else layer.strides
+ padding = [
+ layer.pool_size[0] /
+ 2,
+ layer.pool_size[1] /
+ 2] if layer.padding == 'same' else [
+ 0,
+ 0]
+ return {'pool': pool, 'kernel_h': layer.pool_size[0], 'kernel_w': layer.pool_size[1],
+ 'stride_h': stride[0], 'stride_w': stride[1], 'pad_h': padding[0], 'pad_w': padding[1]}
+
def getRecurrentParam(layer):
- if(not layer.use_bias):
- raise Exception('Only use_bias=True supported for recurrent layers')
- if(keras.activations.serialize(layer.activation) != 'tanh'):
- raise Exception('Only tanh activation supported for recurrent layers')
- if(layer.dropout != 0 or layer.recurrent_dropout != 0):
- raise Exception('Only dropout not supported for recurrent layers')
- return {'num_output': layer.units, 'return_sequences': str(layer.return_sequences).lower() }
-
-# TODO: Update AveragePooling2D when we add maxpooling support
+ if(not layer.use_bias):
+ raise Exception('Only use_bias=True supported for recurrent layers')
+ if(keras.activations.serialize(layer.activation) != 'tanh'):
+ raise Exception('Only tanh activation supported for recurrent layers')
+ if(layer.dropout != 0 or layer.recurrent_dropout != 0):
+ raise Exception('Only dropout not supported for recurrent layers')
+ return {'num_output': layer.units, 'return_sequences': str(
+ layer.return_sequences).lower()}
+
+
+# TODO: Update AveragePooling2D when we add maxpooling support
layerParamMapping = {
- keras.layers.InputLayer: lambda l: \
- {'data_param': {'batch_size': l.batch_size}},
- keras.layers.Dense: lambda l: \
+ keras.layers.InputLayer: lambda l:
+ {'data_param': {'batch_size': l.batch_size}},
+ keras.layers.Dense: lambda l:
{'inner_product_param': {'num_output': l.units}},
- keras.layers.Dropout: lambda l: \
+ keras.layers.Dropout: lambda l:
{'dropout_param': {'dropout_ratio': l.rate}},
- keras.layers.Add: lambda l: \
+ keras.layers.Add: lambda l:
{'eltwise_param': {'operation': 'SUM'}},
- keras.layers.Concatenate: lambda l: \
+ keras.layers.Concatenate: lambda l:
{'concat_param': {'axis': _getCompensatedAxis(l)}},
- keras.layers.Conv2DTranspose: lambda l: \
+ keras.layers.Conv2DTranspose: lambda l:
{'convolution_param': getConvParam(l)},
- keras.layers.UpSampling2D: lambda l: \
+ keras.layers.UpSampling2D: lambda l:
{'upsample_param': getUpSamplingParam(l)},
- keras.layers.Conv2D: lambda l: \
+ keras.layers.Conv2D: lambda l:
{'convolution_param': getConvParam(l)},
- keras.layers.MaxPooling2D: lambda l: \
+ keras.layers.MaxPooling2D: lambda l:
{'pooling_param': getPoolingParam(l, 'MAX')},
- keras.layers.AveragePooling2D: lambda l: \
+ keras.layers.AveragePooling2D: lambda l:
{'pooling_param': getPoolingParam(l, 'AVE')},
- keras.layers.SimpleRNN: lambda l: \
+ keras.layers.SimpleRNN: lambda l:
{'recurrent_param': getRecurrentParam(l)},
- keras.layers.LSTM: lambda l: \
+ keras.layers.LSTM: lambda l:
{'recurrent_param': getRecurrentParam(l)},
- }
+}
+
def _checkIfValid(myList, fn, errorMessage):
- bool_vals = np.array([ fn(elem) for elem in myList])
- unsupported_elems = np.where(bool_vals)[0]
- if len(unsupported_elems) != 0:
- raise ValueError(errorMessage + str(np.array(myList)[unsupported_elems]))
+ bool_vals = np.array([fn(elem) for elem in myList])
+ unsupported_elems = np.where(bool_vals)[0]
+ if len(unsupported_elems) != 0:
+ raise ValueError(errorMessage +
+ str(np.array(myList)[unsupported_elems]))
+
def _transformLayer(layer, batch_size):
- if type(layer) == keras.layers.InputLayer:
- layer.batch_size = batch_size
- return [ layer ]
+ if isinstance(layer, keras.layers.InputLayer):
+ layer.batch_size = batch_size
+ return [layer]
+
def _appendKerasLayers(fileHandle, kerasLayers, batch_size):
- if len(kerasLayers) >= 1:
- transformedLayers = list(chain.from_iterable(imap(lambda layer: _transformLayer(layer, batch_size), kerasLayers)))
- jsonLayers = list(chain.from_iterable(imap(lambda layer: _parseKerasLayer(layer), transformedLayers)))
- parsedLayers = list(chain.from_iterable(imap(lambda layer: _parseJSONObject(layer), jsonLayers)))
- fileHandle.write(''.join(parsedLayers))
- fileHandle.write('\n')
-
+ if len(kerasLayers) >= 1:
+ transformedLayers = list(
+ chain.from_iterable(
+ imap(
+ lambda layer: _transformLayer(
+ layer,
+ batch_size),
+ kerasLayers)))
+ jsonLayers = list(
+ chain.from_iterable(
+ imap(
+ lambda layer: _parseKerasLayer(layer),
+ transformedLayers)))
+ parsedLayers = list(
+ chain.from_iterable(
+ imap(
+ lambda layer: _parseJSONObject(layer),
+ jsonLayers)))
+ fileHandle.write(''.join(parsedLayers))
+ fileHandle.write('\n')
+
+
def lossLayerStr(layerType, bottomLayer):
- return 'layer {\n name: "loss"\n type: "' + layerType + '"\n bottom: "' + bottomLayer + '"\n bottom: "label"\n top: "loss"\n}\n'
-
+ return 'layer {\n name: "loss"\n type: "' + layerType + \
+ '"\n bottom: "' + bottomLayer + '"\n bottom: "label"\n top: "loss"\n}\n'
+
+
def _appendKerasLayerWithoutActivation(fileHandle, layer, batch_size):
- if type(layer) != keras.layers.Activation:
- lastLayerActivation = layer.activation
- layer.activation = keras.activations.linear
- _appendKerasLayers(fileHandle, [layer], batch_size)
- layer.activation = lastLayerActivation
+ if not isinstance(layer, keras.layers.Activation):
+ lastLayerActivation = layer.activation
+ layer.activation = keras.activations.linear
+ _appendKerasLayers(fileHandle, [layer], batch_size)
+ layer.activation = lastLayerActivation
+
def _getExactlyOneBottomLayer(layer):
- bottomLayers = _getBottomLayers(layer)
- if len(bottomLayers) != 1:
- raise Exception('Expected only one bottom layer for ' + str(layer.name) + ', but found ' + str(bottomLayers))
- return bottomLayers[0]
+ bottomLayers = _getBottomLayers(layer)
+ if len(bottomLayers) != 1:
+ raise Exception('Expected only one bottom layer for ' +
+ str(layer.name) + ', but found ' + str(bottomLayers))
+ return bottomLayers[0]
+
def _isMeanSquaredError(loss):
- return loss == 'mean_squared_error' or loss == 'mse' or loss == 'MSE'
-
-def convertKerasToCaffeNetwork(kerasModel, outCaffeNetworkFilePath, batch_size):
- _checkIfValid(kerasModel.layers, lambda layer: False if type(layer) in supportedLayers else True, 'Unsupported Layers:')
- with open(outCaffeNetworkFilePath, 'w') as f:
- # Write the parsed layers for all but the last layer
- _appendKerasLayers(f, kerasModel.layers[:-1], batch_size)
- # Now process the last layer with loss
- lastLayer = kerasModel.layers[-1]
- if _isMeanSquaredError(kerasModel.loss):
- _appendKerasLayers(f, [ lastLayer ], batch_size)
- f.write(lossLayerStr('EuclideanLoss', lastLayer.name))
- elif kerasModel.loss == 'categorical_crossentropy':
- _appendKerasLayerWithoutActivation(f, lastLayer, batch_size)
- bottomLayer = _getExactlyOneBottomLayer(lastLayer) if type(lastLayer) == keras.layers.Activation else lastLayer.name
- lastLayerActivation = str(keras.activations.serialize(lastLayer.activation))
- if lastLayerActivation == 'softmax' and kerasModel.loss == 'categorical_crossentropy':
- f.write(lossLayerStr('SoftmaxWithLoss', bottomLayer))
- else:
- raise Exception('Unsupported loss layer ' + str(kerasModel.loss) + ' (where last layer activation ' + lastLayerActivation + ').')
- else:
- raise Exception('Unsupported loss layer ' + str(kerasModel.loss) + ' (where last layer activation ' + lastLayerActivation + ').')
+ return loss == 'mean_squared_error' or loss == 'mse' or loss == 'MSE'
+
+
+def convertKerasToCaffeNetwork(
+ kerasModel, outCaffeNetworkFilePath, batch_size):
+ _checkIfValid(kerasModel.layers, lambda layer: False if type(
+ layer) in supportedLayers else True, 'Unsupported Layers:')
+ with open(outCaffeNetworkFilePath, 'w') as f:
+ # Write the parsed layers for all but the last layer
+ _appendKerasLayers(f, kerasModel.layers[:-1], batch_size)
+ # Now process the last layer with loss
+ lastLayer = kerasModel.layers[-1]
+ if _isMeanSquaredError(kerasModel.loss):
+ _appendKerasLayers(f, [lastLayer], batch_size)
+ f.write(lossLayerStr('EuclideanLoss', lastLayer.name))
+ elif kerasModel.loss == 'categorical_crossentropy':
+ _appendKerasLayerWithoutActivation(f, lastLayer, batch_size)
+ bottomLayer = _getExactlyOneBottomLayer(lastLayer) if isinstance(
+ lastLayer, keras.layers.Activation) else lastLayer.name
+ lastLayerActivation = str(
+ keras.activations.serialize(
+ lastLayer.activation))
+ if lastLayerActivation == 'softmax' and kerasModel.loss == 'categorical_crossentropy':
+ f.write(lossLayerStr('SoftmaxWithLoss', bottomLayer))
+ else:
+ raise Exception('Unsupported loss layer ' +
+ str(kerasModel.loss) +
+ ' (where last layer activation ' +
+ lastLayerActivation +
+ ').')
+ else:
+ raise Exception('Unsupported loss layer ' +
+ str(kerasModel.loss) +
+ ' (where last layer activation ' +
+ lastLayerActivation +
+ ').')
def getNumPyMatrixFromKerasWeight(param):
- x = np.array(param)
- if len(x.shape) > 2:
- x = x.transpose(3, 2, 0, 1)
- return x.reshape(x.shape[0], -1)
- elif len(x.shape) == 1:
- return np.matrix(param).transpose()
- else:
- return x
+ x = np.array(param)
+ if len(x.shape) > 2:
+ x = x.transpose(3, 2, 0, 1)
+ return x.reshape(x.shape[0], -1)
+ elif len(x.shape) == 1:
+ return np.matrix(param).transpose()
+ else:
+ return x
defaultSolver = """
solver_mode: CPU
"""
+
def evaluateValue(val):
- if type(val) == int or type(val) == float:
- return float(val)
- else:
- return K.eval(val)
-
-def convertKerasToCaffeSolver(kerasModel, caffeNetworkFilePath, outCaffeSolverFilePath, max_iter, test_iter, test_interval, display, lr_policy, weight_decay, regularization_type):
- if type(kerasModel.optimizer) == keras.optimizers.SGD:
- solver = 'type: "Nesterov"\n' if kerasModel.optimizer.nesterov else 'type: "SGD"\n'
- elif type(kerasModel.optimizer) == keras.optimizers.Adagrad:
- solver = 'type: "Adagrad"\n'
- elif type(kerasModel.optimizer) == keras.optimizers.Adam:
- solver = 'type: "Adam"\n'
- else:
- raise Exception('Only sgd (with/without momentum/nesterov), Adam and Adagrad supported.')
- base_lr = evaluateValue(kerasModel.optimizer.lr) if hasattr(kerasModel.optimizer, 'lr') else 0.01
- gamma = evaluateValue(kerasModel.optimizer.decay) if hasattr(kerasModel.optimizer, 'decay') else 0.0
- with open(outCaffeSolverFilePath, 'w') as f:
- f.write('net: "' + caffeNetworkFilePath + '"\n')
- f.write(defaultSolver)
- f.write(solver)
- f.write('lr_policy: "' + lr_policy + '"\n')
- f.write('regularization_type: "' + str(regularization_type) + '"\n')
- f.write('weight_decay: ' + str(weight_decay) + '\n')
- f.write('max_iter: ' + str(max_iter) + '\ntest_iter: ' + str(test_iter) + '\ntest_interval: ' + str(test_interval) + '\n')
- f.write('display: ' + str(display) + '\n')
- f.write('base_lr: ' + str(base_lr) + '\n')
- f.write('gamma: ' + str(gamma) + '\n')
- if type(kerasModel.optimizer) == keras.optimizers.SGD:
- momentum = evaluateValue(kerasModel.optimizer.momentum) if hasattr(kerasModel.optimizer, 'momentum') else 0.0
- f.write('momentum: ' + str(momentum) + '\n')
- elif type(kerasModel.optimizer) == keras.optimizers.Adam:
- momentum = evaluateValue(kerasModel.optimizer.beta_1) if hasattr(kerasModel.optimizer, 'beta_1') else 0.9
- momentum2 = evaluateValue(kerasModel.optimizer.beta_2) if hasattr(kerasModel.optimizer, 'beta_2') else 0.999
- delta = evaluateValue(kerasModel.optimizer.epsilon) if hasattr(kerasModel.optimizer, 'epsilon') else 1e-8
- f.write('momentum: ' + str(momentum) + '\n')
- f.write('momentum2: ' + str(momentum2) + '\n')
- f.write('delta: ' + str(delta) + '\n')
- elif type(kerasModel.optimizer) == keras.optimizers.Adagrad:
- delta = evaluateValue(kerasModel.optimizer.epsilon) if hasattr(kerasModel.optimizer, 'epsilon') else 1e-8
- f.write('delta: ' + str(delta) + '\n')
- else:
- raise Exception('Only sgd (with/without momentum/nesterov), Adam and Adagrad supported.')
+ if isinstance(val, int) or isinstance(val, float):
+ return float(val)
+ else:
+ return K.eval(val)
+
+
+def convertKerasToCaffeSolver(kerasModel, caffeNetworkFilePath, outCaffeSolverFilePath,
+ max_iter, test_iter, test_interval, display, lr_policy, weight_decay, regularization_type):
+ if isinstance(kerasModel.optimizer, keras.optimizers.SGD):
+ solver = 'type: "Nesterov"\n' if kerasModel.optimizer.nesterov else 'type: "SGD"\n'
+ elif isinstance(kerasModel.optimizer, keras.optimizers.Adagrad):
+ solver = 'type: "Adagrad"\n'
+ elif isinstance(kerasModel.optimizer, keras.optimizers.Adam):
+ solver = 'type: "Adam"\n'
+ else:
+ raise Exception(
+ 'Only sgd (with/without momentum/nesterov), Adam and Adagrad supported.')
+ base_lr = evaluateValue(
+ kerasModel.optimizer.lr) if hasattr(
+ kerasModel.optimizer,
+ 'lr') else 0.01
+ gamma = evaluateValue(
+ kerasModel.optimizer.decay) if hasattr(
+ kerasModel.optimizer,
+ 'decay') else 0.0
+ with open(outCaffeSolverFilePath, 'w') as f:
+ f.write('net: "' + caffeNetworkFilePath + '"\n')
+ f.write(defaultSolver)
+ f.write(solver)
+ f.write('lr_policy: "' + lr_policy + '"\n')
+ f.write('regularization_type: "' + str(regularization_type) + '"\n')
+ f.write('weight_decay: ' + str(weight_decay) + '\n')
+ f.write(
+ 'max_iter: ' +
+ str(max_iter) +
+ '\ntest_iter: ' +
+ str(test_iter) +
+ '\ntest_interval: ' +
+ str(test_interval) +
+ '\n')
+ f.write('display: ' + str(display) + '\n')
+ f.write('base_lr: ' + str(base_lr) + '\n')
+ f.write('gamma: ' + str(gamma) + '\n')
+ if isinstance(kerasModel.optimizer, keras.optimizers.SGD):
+ momentum = evaluateValue(
+ kerasModel.optimizer.momentum) if hasattr(
+ kerasModel.optimizer,
+ 'momentum') else 0.0
+ f.write('momentum: ' + str(momentum) + '\n')
+ elif isinstance(kerasModel.optimizer, keras.optimizers.Adam):
+ momentum = evaluateValue(
+ kerasModel.optimizer.beta_1) if hasattr(
+ kerasModel.optimizer,
+ 'beta_1') else 0.9
+ momentum2 = evaluateValue(
+ kerasModel.optimizer.beta_2) if hasattr(
+ kerasModel.optimizer,
+ 'beta_2') else 0.999
+ delta = evaluateValue(
+ kerasModel.optimizer.epsilon) if hasattr(
+ kerasModel.optimizer,
+ 'epsilon') else 1e-8
+ f.write('momentum: ' + str(momentum) + '\n')
+ f.write('momentum2: ' + str(momentum2) + '\n')
+ f.write('delta: ' + str(delta) + '\n')
+ elif isinstance(kerasModel.optimizer, keras.optimizers.Adagrad):
+ delta = evaluateValue(
+ kerasModel.optimizer.epsilon) if hasattr(
+ kerasModel.optimizer,
+ 'epsilon') else 1e-8
+ f.write('delta: ' + str(delta) + '\n')
+ else:
+ raise Exception(
+ 'Only sgd (with/without momentum/nesterov), Adam and Adagrad supported.')
def getInputMatrices(layer):
- if type(layer) == keras.layers.LSTM or type(layer) == keras.layers.SimpleRNN:
- weights = layer.get_weights()
- return [np.vstack((weights[0], weights[1])), np.matrix(weights[2]) ]
- else:
- return [ getNumPyMatrixFromKerasWeight(param) for param in layer.get_weights() ]
+ if isinstance(layer, keras.layers.LSTM) or isinstance(
+ layer, keras.layers.SimpleRNN):
+ weights = layer.get_weights()
+ return [np.vstack((weights[0], weights[1])), np.matrix(weights[2])]
+ else:
+ return [getNumPyMatrixFromKerasWeight(
+ param) for param in layer.get_weights()]
+
def convertKerasToSystemMLModel(spark, kerasModel, outDirectory):
- _checkIfValid(kerasModel.layers, lambda layer: False if len(layer.get_weights()) <= 4 or len(layer.get_weights()) != 3 else True, 'Unsupported number of weights:')
- layers = [layer for layer in kerasModel.layers if len(layer.get_weights()) > 0]
- sc = spark._sc
- biasToTranspose = [ keras.layers.Dense ]
- dmlLines = []
- script_java = sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dml('')
- for layer in layers:
- inputMatrices = getInputMatrices(layer)
- potentialVar = [ layer.name + '_weight', layer.name + '_bias', layer.name + '_1_weight', layer.name + '_1_bias' ]
- for i in range(len(inputMatrices)):
- dmlLines = dmlLines + [ 'write(' + potentialVar[i] + ', "' + outDirectory + '/' + potentialVar[i] + '.mtx", format="binary");\n' ]
- mat = inputMatrices[i].transpose() if (i == 1 and type(layer) in biasToTranspose) else inputMatrices[i]
- py4j.java_gateway.get_method(script_java, "in")(potentialVar[i], convertToMatrixBlock(sc, mat))
- script_java.setScriptString(''.join(dmlLines))
- ml = sc._jvm.org.apache.sysml.api.mlcontext.MLContext(sc._jsc)
- ml.execute(script_java)
+ _checkIfValid(
+ kerasModel.layers,
+ lambda layer: False if len(
+ layer.get_weights()) <= 4 or len(
+ layer.get_weights()) != 3 else True,
+ 'Unsupported number of weights:')
+ layers = [
+ layer for layer in kerasModel.layers if len(
+ layer.get_weights()) > 0]
+ sc = spark._sc
+ biasToTranspose = [keras.layers.Dense]
+ dmlLines = []
+ script_java = sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dml('')
+ for layer in layers:
+ inputMatrices = getInputMatrices(layer)
+ potentialVar = [
+ layer.name + '_weight',
+ layer.name + '_bias',
+ layer.name + '_1_weight',
+ layer.name + '_1_bias']
+ for i in range(len(inputMatrices)):
+ dmlLines = dmlLines + \
+ ['write(' + potentialVar[i] + ', "' + outDirectory +
+ '/' + potentialVar[i] + '.mtx", format="binary");\n']
+ mat = inputMatrices[i].transpose() if (
+ i == 1 and type(layer) in biasToTranspose) else inputMatrices[i]
+ py4j.java_gateway.get_method(script_java, "in")(
+ potentialVar[i], convertToMatrixBlock(sc, mat))
+ script_java.setScriptString(''.join(dmlLines))
+ ml = sc._jvm.org.apache.sysml.api.mlcontext.MLContext(sc._jsc)
+ ml.execute(script_java)
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/random/__init__.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/random/__init__.py b/src/main/python/systemml/random/__init__.py
index 65f6bbd..68a74b8 100644
--- a/src/main/python/systemml/random/__init__.py
+++ b/src/main/python/systemml/random/__init__.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,7 +17,7 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
"""
========================
@@ -38,4 +38,4 @@ uniform Uniform distribution.
from .sampling import *
-__all__ = sampling.__all__
\ No newline at end of file
+__all__ = sampling.__all__
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/random/sampling.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/random/sampling.py b/src/main/python/systemml/random/sampling.py
index d320536..ab74250 100644
--- a/src/main/python/systemml/random/sampling.py
+++ b/src/main/python/systemml/random/sampling.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,7 +17,7 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
__all__ = ['normal', 'uniform', 'poisson']
@@ -27,6 +27,7 @@ from ..defmatrix import *
# This helps to provide dml containing output ID in constructSamplingNode
OUTPUT_ID = '$$OutputID$$'
+
def constructSamplingNode(inputs, dml):
"""
Convenient utility to create an intermediate of AST.
@@ -38,31 +39,35 @@ def constructSamplingNode(inputs, dml):
"""
dmlOp = DMLOp(inputs)
out = matrix(None, op=dmlOp)
- dmlOp.dml = [out.ID if x==OUTPUT_ID else x for x in dml]
+ dmlOp.dml = [out.ID if x == OUTPUT_ID else x for x in dml]
return out
+
INPUTS = []
+
+
def asStr(arg):
"""
Internal use only: Convenient utility to update inputs and return appropriate string value
"""
if isinstance(arg, matrix):
- INPUTS = INPUTS + [ arg ]
+ INPUTS = INPUTS + [arg]
return arg.ID
else:
return str(arg)
-
-def normal(loc=0.0, scale=1.0, size=(1,1), sparsity=1.0):
+
+
+def normal(loc=0.0, scale=1.0, size=(1, 1), sparsity=1.0):
"""
Draw random samples from a normal (Gaussian) distribution.
-
+
Parameters
----------
loc: Mean ("centre") of the distribution.
scale: Standard deviation (spread or "width") of the distribution.
size: Output shape (only tuple of length 2, i.e. (m, n), supported).
sparsity: Sparsity (between 0.0 and 1.0).
-
+
Examples
--------
@@ -75,7 +80,7 @@ def normal(loc=0.0, scale=1.0, size=(1,1), sparsity=1.0):
array([[ 3.48857226, 6.17261819, 2.51167259],
[ 3.60506708, -1.90266305, 3.97601633],
[ 3.62245706, 5.9430881 , 2.53070413]])
-
+
"""
if len(size) != 2:
raise TypeError('Incorrect type for size. Expected tuple of length 2')
@@ -86,22 +91,24 @@ def normal(loc=0.0, scale=1.0, size=(1,1), sparsity=1.0):
scale = asStr(scale)
sparsity = asStr(sparsity)
# loc + scale*standard normal
- return constructSamplingNode(INPUTS, [OUTPUT_ID, ' = ', loc,' + ', scale,' * random.normal(', rows, ',', cols, ',', sparsity, ')\n'])
+ return constructSamplingNode(INPUTS, [
+ OUTPUT_ID, ' = ', loc, ' + ', scale, ' * random.normal(', rows, ',', cols, ',', sparsity, ')\n'])
+
-def uniform(low=0.0, high=1.0, size=(1,1), sparsity=1.0):
+def uniform(low=0.0, high=1.0, size=(1, 1), sparsity=1.0):
"""
Draw samples from a uniform distribution.
-
+
Parameters
----------
low: Lower boundary of the output interval.
high: Upper boundary of the output interval.
- size: Output shape (only tuple of length 2, i.e. (m, n), supported).
+ size: Output shape (only tuple of length 2, i.e. (m, n), supported).
sparsity: Sparsity (between 0.0 and 1.0).
Examples
--------
-
+
>>> import systemml as sml
>>> import numpy as np
>>> sml.setSparkContext(sc)
@@ -121,22 +128,24 @@ def uniform(low=0.0, high=1.0, size=(1,1), sparsity=1.0):
low = asStr(low)
high = asStr(high)
sparsity = asStr(sparsity)
- return constructSamplingNode(INPUTS, [OUTPUT_ID, ' = random.uniform(', rows, ',', cols, ',', sparsity, ',', low, ',', high, ')\n'])
+ return constructSamplingNode(INPUTS, [
+ OUTPUT_ID, ' = random.uniform(', rows, ',', cols, ',', sparsity, ',', low, ',', high, ')\n'])
-def poisson(lam=1.0, size=(1,1), sparsity=1.0):
+
+def poisson(lam=1.0, size=(1, 1), sparsity=1.0):
"""
Draw samples from a Poisson distribution.
-
+
Parameters
----------
lam: Expectation of interval, should be > 0.
- size: Output shape (only tuple of length 2, i.e. (m, n), supported).
+ size: Output shape (only tuple of length 2, i.e. (m, n), supported).
sparsity: Sparsity (between 0.0 and 1.0).
-
-
+
+
Examples
--------
-
+
>>> import systemml as sml
>>> import numpy as np
>>> sml.setSparkContext(sc)
@@ -146,7 +155,7 @@ def poisson(lam=1.0, size=(1,1), sparsity=1.0):
array([[ 1., 0., 2.],
[ 1., 0., 0.],
[ 0., 0., 0.]])
-
+
"""
if len(size) != 2:
raise TypeError('Incorrect type for size. Expected tuple of length 2')
@@ -155,4 +164,5 @@ def poisson(lam=1.0, size=(1,1), sparsity=1.0):
cols = asStr(size[1])
lam = asStr(lam)
sparsity = asStr(sparsity)
- return constructSamplingNode(INPUTS, [OUTPUT_ID, ' = random.poisson(', rows, ',', cols, ',', sparsity, ',', lam, ')\n'])
+ return constructSamplingNode(INPUTS, [
+ OUTPUT_ID, ' = random.poisson(', rows, ',', cols, ',', sparsity, ',', lam, ')\n'])
[3/3] systemml git commit: [MINOR] Refactored the Python code to
match PEP8 guidelines and bugfix in Keras2DML for Python 3.5
Posted by ni...@apache.org.
[MINOR] Refactored the Python code to match PEP8 guidelines and bugfix in Keras2DML for Python 3.5
Project: http://git-wip-us.apache.org/repos/asf/systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/systemml/commit/9e7ee19a
Tree: http://git-wip-us.apache.org/repos/asf/systemml/tree/9e7ee19a
Diff: http://git-wip-us.apache.org/repos/asf/systemml/diff/9e7ee19a
Branch: refs/heads/master
Commit: 9e7ee19a45102f7cbb37507da25b1ba0641868fd
Parents: ddc3672
Author: Niketan Pansare <np...@us.ibm.com>
Authored: Mon Aug 27 17:26:55 2018 -0700
Committer: Niketan Pansare <np...@us.ibm.com>
Committed: Mon Aug 27 17:34:17 2018 -0700
----------------------------------------------------------------------
src/main/python/systemml/__init__.py | 4 +-
src/main/python/systemml/classloader.py | 90 ++-
src/main/python/systemml/converters.py | 170 ++++--
src/main/python/systemml/defmatrix.py | 565 ++++++++++++-------
src/main/python/systemml/mlcontext.py | 229 +++++---
src/main/python/systemml/mllearn/__init__.py | 4 +-
src/main/python/systemml/mllearn/estimators.py | 357 ++++++++----
src/main/python/systemml/mllearn/keras2caffe.py | 553 +++++++++++-------
src/main/python/systemml/random/__init__.py | 6 +-
src/main/python/systemml/random/sampling.py | 56 +-
10 files changed, 1332 insertions(+), 702 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/__init__.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/__init__.py b/src/main/python/systemml/__init__.py
index 04d521bb..74c8e50 100644
--- a/src/main/python/systemml/__init__.py
+++ b/src/main/python/systemml/__init__.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,7 +17,7 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
from .mlcontext import *
from .defmatrix import *
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/classloader.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/classloader.py b/src/main/python/systemml/classloader.py
index 229664d..1efbdd1 100644
--- a/src/main/python/systemml/classloader.py
+++ b/src/main/python/systemml/classloader.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,14 +17,21 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
-__all__ = ['createJavaObject', 'jvm_stdout', 'default_jvm_stdout', 'default_jvm_stdout_parallel_flush', 'set_default_jvm_stdout', 'get_spark_context' ]
+__all__ = [
+ 'createJavaObject',
+ 'jvm_stdout',
+ 'default_jvm_stdout',
+ 'default_jvm_stdout_parallel_flush',
+ 'set_default_jvm_stdout',
+ 'get_spark_context']
import os
import numpy as np
import pandas as pd
-import threading, time
+import threading
+import time
try:
import py4j.java_gateway
@@ -32,9 +39,12 @@ try:
from pyspark import SparkContext
from pyspark.sql import SparkSession
except ImportError:
- raise ImportError('Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
+ raise ImportError(
+ 'Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
_loadedSystemML = False
+
+
def get_spark_context():
"""
Internal method to get already initialized SparkContext. Developers should always use
@@ -54,10 +64,13 @@ def get_spark_context():
return sc
else:
raise Exception('Expected spark context to be created.')
-
+
+
_in_jvm_stdout = False
default_jvm_stdout = True
default_jvm_stdout_parallel_flush = True
+
+
def set_default_jvm_stdout(enable, parallel_flush=True):
"""
This is useful utility method to get the output of the driver JVM from within a Jupyter notebook
@@ -66,18 +79,20 @@ def set_default_jvm_stdout(enable, parallel_flush=True):
----------
enable: boolean
Should flush the stdout by default when mlcontext.execute is invoked
-
+
parallel_flush: boolean
Should flush the stdout in parallel
"""
global default_jvm_stdout, default_jvm_stdout_parallel_flush
default_jvm_stdout = enable
default_jvm_stdout_parallel_flush = parallel_flush
-
+
# This is useful utility class to get the output of the driver JVM from within a Jupyter notebook
# Example usage:
# with jvm_stdout():
# ml.execute(script)
+
+
class jvm_stdout(object):
"""
This is useful utility class to get the output of the driver JVM from within a Jupyter notebook
@@ -87,24 +102,25 @@ class jvm_stdout(object):
parallel_flush: boolean
Should flush the stdout in parallel
"""
+
def __init__(self, parallel_flush=False):
self.util = get_spark_context()._jvm.org.apache.sysml.api.ml.Utils()
self.parallel_flush = parallel_flush
self.t = threading.Thread(target=self.flush_stdout)
self.stop = False
-
+
def flush_stdout(self):
- while not self.stop:
- time.sleep(1) # flush stdout every 1 second
+ while not self.stop:
+ time.sleep(1) # flush stdout every 1 second
str = self.util.flushStdOut()
if str != '':
str = str[:-1] if str.endswith('\n') else str
print(str)
-
+
def __enter__(self):
global _in_jvm_stdout
if _in_jvm_stdout:
- # Allow for nested jvm_stdout
+ # Allow for nested jvm_stdout
self.donotRedirect = True
else:
self.donotRedirect = False
@@ -124,32 +140,41 @@ class jvm_stdout(object):
_initializedSparkSession = False
+
+
def _createJavaObject(sc, obj_type):
# -----------------------------------------------------------------------------------
# Avoids race condition between locking of metastore_db of Scala SparkSession and PySpark SparkSession.
- # This is done at toDF() rather than import level to avoid creation of SparkSession in worker processes.
+ # This is done at toDF() rather than import level to avoid creation of
+ # SparkSession in worker processes.
global _initializedSparkSession
if not _initializedSparkSession:
_initializedSparkSession = True
- SparkSession.builder.getOrCreate().createDataFrame(pd.DataFrame(np.array([[1,2],[3,4]])))
+ SparkSession.builder.getOrCreate().createDataFrame(
+ pd.DataFrame(np.array([[1, 2], [3, 4]])))
# -----------------------------------------------------------------------------------
if obj_type == 'mlcontext':
return sc._jvm.org.apache.sysml.api.mlcontext.MLContext(sc._jsc)
elif obj_type == 'dummy':
return sc._jvm.org.apache.sysml.utils.SystemMLLoaderUtils()
else:
- raise ValueError('Incorrect usage: supported values: mlcontext or dummy')
+ raise ValueError(
+ 'Incorrect usage: supported values: mlcontext or dummy')
+
def _getJarFileNames(sc):
- import imp, fnmatch
+ import imp
+ import fnmatch
jar_file_name = '_ignore.jar'
java_dir = os.path.join(imp.find_module("systemml")[1], "systemml-java")
jar_file_names = []
for file in os.listdir(java_dir):
- if fnmatch.fnmatch(file, 'systemml-*-SNAPSHOT.jar') or fnmatch.fnmatch(file, 'systemml-*.jar'):
- jar_file_names = jar_file_names + [ os.path.join(java_dir, file) ]
+ if fnmatch.fnmatch(
+ file, 'systemml-*-SNAPSHOT.jar') or fnmatch.fnmatch(file, 'systemml-*.jar'):
+ jar_file_names = jar_file_names + [os.path.join(java_dir, file)]
return jar_file_names
+
def _getLoaderInstance(sc, jar_file_name, className, hint):
err_msg = 'Unable to load systemml-*.jar into current pyspark session.'
if os.path.isfile(jar_file_name):
@@ -158,16 +183,21 @@ def _getLoaderInstance(sc, jar_file_name, className, hint):
url_class = sc._jvm.java.net.URL
jar_file_url_arr = sc._gateway.new_array(url_class, 1)
jar_file_url_arr[0] = jar_file_url
- url_class_loader = sc._jvm.java.net.URLClassLoader(jar_file_url_arr, sc._jsc.getClass().getClassLoader())
+ url_class_loader = sc._jvm.java.net.URLClassLoader(
+ jar_file_url_arr, sc._jsc.getClass().getClassLoader())
c1 = sc._jvm.java.lang.Class.forName(className, True, url_class_loader)
return c1.newInstance()
else:
- raise ImportError(err_msg + ' Hint: Download the jar from http://systemml.apache.org/download and ' + hint )
+ raise ImportError(
+ err_msg +
+ ' Hint: Download the jar from http://systemml.apache.org/download and ' +
+ hint)
+
def createJavaObject(sc, obj_type):
"""
Performs appropriate check if SystemML.jar is available and returns the handle to MLContext object on JVM
-
+
Parameters
----------
sc: SparkContext
@@ -182,13 +212,23 @@ def createJavaObject(sc, obj_type):
hint = 'Provide the following argument to pyspark: --driver-class-path '
jar_file_names = _getJarFileNames(sc)
if len(jar_file_names) != 2:
- raise ImportError('Expected only systemml and systemml-extra jars, but found ' + str(jar_file_names))
+ raise ImportError(
+ 'Expected only systemml and systemml-extra jars, but found ' +
+ str(jar_file_names))
for jar_file_name in jar_file_names:
if 'extra' in jar_file_name:
- x = _getLoaderInstance(sc, jar_file_name, 'org.apache.sysml.api.dl.Caffe2DMLLoader', hint + 'systemml-*-extra.jar')
+ x = _getLoaderInstance(
+ sc,
+ jar_file_name,
+ 'org.apache.sysml.api.dl.Caffe2DMLLoader',
+ hint + 'systemml-*-extra.jar')
x.loadCaffe2DML(jar_file_name)
else:
- x = _getLoaderInstance(sc, jar_file_name, 'org.apache.sysml.utils.SystemMLLoaderUtils', hint + 'systemml-*.jar')
+ x = _getLoaderInstance(
+ sc,
+ jar_file_name,
+ 'org.apache.sysml.utils.SystemMLLoaderUtils',
+ hint + 'systemml-*.jar')
x.loadSystemML(jar_file_name)
try:
ret = _createJavaObject(sc, obj_type)
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/converters.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/converters.py b/src/main/python/systemml/converters.py
index aaa9f88..5954a30 100644
--- a/src/main/python/systemml/converters.py
+++ b/src/main/python/systemml/converters.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,9 +17,19 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
-
-__all__ = [ 'getNumCols', 'convertToMatrixBlock', 'convert_caffemodel', 'convert_lmdb_to_jpeg', 'convertToNumPyArr', 'convertToPandasDF', 'SUPPORTED_TYPES' , 'convertToLabeledDF', 'convertImageToNumPyArr', 'getDatasetMean']
+# -------------------------------------------------------------
+
+__all__ = [
+ 'getNumCols',
+ 'convertToMatrixBlock',
+ 'convert_caffemodel',
+ 'convert_lmdb_to_jpeg',
+ 'convertToNumPyArr',
+ 'convertToPandasDF',
+ 'SUPPORTED_TYPES',
+ 'convertToLabeledDF',
+ 'convertImageToNumPyArr',
+ 'getDatasetMean']
import numpy as np
import pandas as pd
@@ -32,7 +42,8 @@ from .classloader import *
SUPPORTED_TYPES = (np.ndarray, pd.DataFrame, spmatrix)
-DATASET_MEAN = {'VGG_ILSVRC_19_2014':[103.939, 116.779, 123.68]}
+DATASET_MEAN = {'VGG_ILSVRC_19_2014': [103.939, 116.779, 123.68]}
+
def getNumCols(numPyArr):
if numPyArr.ndim == 1:
@@ -40,9 +51,11 @@ def getNumCols(numPyArr):
else:
return numPyArr.shape[1]
+
def get_pretty_str(key, value):
return '\t"' + key + '": ' + str(value) + ',\n'
+
def save_tensor_csv(tensor, file_path, shouldTranspose):
w = w.reshape(w.shape[0], -1)
if shouldTranspose:
@@ -53,9 +66,12 @@ def save_tensor_csv(tensor, file_path, shouldTranspose):
file.write(get_pretty_str('rows', w.shape[0]))
file.write(get_pretty_str('cols', w.shape[1]))
file.write(get_pretty_str('nnz', np.count_nonzero(w)))
- file.write('\t"format": "csv",\n\t"description": {\n\t\t"author": "SystemML"\n\t}\n}\n')
+ file.write(
+ '\t"format": "csv",\n\t"description": {\n\t\t"author": "SystemML"\n\t}\n}\n')
-def convert_caffemodel(sc, deploy_file, caffemodel_file, output_dir, format="binary", is_caffe_installed=False):
+
+def convert_caffemodel(sc, deploy_file, caffemodel_file,
+ output_dir, format="binary", is_caffe_installed=False):
"""
Saves the weights and bias in the caffemodel file to output_dir in the specified format.
This method does not requires caffe to be installed.
@@ -82,7 +98,10 @@ def convert_caffemodel(sc, deploy_file, caffemodel_file, output_dir, format="bin
"""
if is_caffe_installed:
if format != 'csv':
- raise ValueError('The format ' + str(format) + ' is not supported when caffe is installed. Hint: Please specify format=csv')
+ raise ValueError(
+ 'The format ' +
+ str(format) +
+ ' is not supported when caffe is installed. Hint: Please specify format=csv')
import caffe
net = caffe.Net(deploy_file, caffemodel_file, caffe.TEST)
for layerName in net.params.keys():
@@ -91,21 +110,48 @@ def convert_caffemodel(sc, deploy_file, caffemodel_file, output_dir, format="bin
continue
elif num_parameters == 2:
# Weights and Biases
- layerType = net.layers[list(net._layer_names).index(layerName)].type
+ layerType = net.layers[list(
+ net._layer_names).index(layerName)].type
shouldTranspose = True if layerType == 'InnerProduct' else False
- save_tensor_csv(net.params[layerName][0].data, os.path.join(output_dir, layerName + '_weight.mtx'), shouldTranspose)
- save_tensor_csv(net.params[layerName][1].data, os.path.join(output_dir, layerName + '_bias.mtx'), shouldTranspose)
+ save_tensor_csv(
+ net.params[layerName][0].data,
+ os.path.join(
+ output_dir,
+ layerName +
+ '_weight.mtx'),
+ shouldTranspose)
+ save_tensor_csv(
+ net.params[layerName][1].data,
+ os.path.join(
+ output_dir,
+ layerName +
+ '_bias.mtx'),
+ shouldTranspose)
elif num_parameters == 1:
# Only Weight
- layerType = net.layers[list(net._layer_names).index(layerName)].type
+ layerType = net.layers[list(
+ net._layer_names).index(layerName)].type
shouldTranspose = True if layerType == 'InnerProduct' else False
- save_tensor_csv(net.params[layerName][0].data, os.path.join(output_dir, layerName + '_weight.mtx'), shouldTranspose)
+ save_tensor_csv(
+ net.params[layerName][0].data,
+ os.path.join(
+ output_dir,
+ layerName +
+ '_weight.mtx'),
+ shouldTranspose)
else:
- raise ValueError('Unsupported number of parameters:' + str(num_parameters))
+ raise ValueError(
+ 'Unsupported number of parameters:' +
+ str(num_parameters))
else:
createJavaObject(sc, 'dummy')
utilObj = sc._jvm.org.apache.sysml.api.dl.Utils()
- utilObj.saveCaffeModelFile(sc._jsc, deploy_file, caffemodel_file, output_dir, format)
+ utilObj.saveCaffeModelFile(
+ sc._jsc,
+ deploy_file,
+ caffemodel_file,
+ output_dir,
+ format)
def convert_lmdb_to_jpeg(lmdb_img_file, output_dir):
@@ -121,7 +167,9 @@ def convert_lmdb_to_jpeg(lmdb_img_file, output_dir):
output_dir: string
Output directory for images (local filesystem)
"""
- import lmdb, caffe, cv2
+ import lmdb
+ import caffe
+ import cv2
lmdb_cursor = lmdb.open(lmdb_file, readonly=True).begin().cursor()
datum = caffe.proto.caffe_pb2.Datum()
i = 1
@@ -129,7 +177,7 @@ def convert_lmdb_to_jpeg(lmdb_img_file, output_dir):
datum.ParseFromString(value)
data = caffe.io.datum_to_array(datum)
output_file_path = os.path.join(output_dir, 'file_' + str(i) + '.jpg')
- image = np.transpose(data, (1,2,0)) # CxHxW to HxWxC in cv2
+ image = np.transpose(data, (1, 2, 0)) # CxHxW to HxWxC in cv2
cv2.imwrite(output_file_path, image)
i = i + 1
@@ -153,8 +201,9 @@ def convertToLabeledDF(sparkSession, X, y=None):
else:
return out.select('features')
+
def _convertSPMatrixToMB(sc, src):
- src = coo_matrix(src, dtype=np.float64)
+ src = coo_matrix(src, dtype=np.float64)
numRows = src.shape[0]
numCols = src.shape[1]
data = src.data
@@ -165,7 +214,9 @@ def _convertSPMatrixToMB(sc, src):
buf2 = bytearray(row.tostring())
buf3 = bytearray(col.tostring())
createJavaObject(sc, 'dummy')
- return sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.convertSciPyCOOToMB(buf1, buf2, buf3, numRows, numCols, nnz)
+ return sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.convertSciPyCOOToMB(
+ buf1, buf2, buf3, numRows, numCols, nnz)
+
def _convertDenseMatrixToMB(sc, src):
numCols = getNumCols(src)
@@ -173,15 +224,25 @@ def _convertDenseMatrixToMB(sc, src):
arr = src.ravel().astype(np.float64)
buf = bytearray(arr.tostring())
createJavaObject(sc, 'dummy')
- return sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.convertPy4JArrayToMB(buf, numRows, numCols)
+ return sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.convertPy4JArrayToMB(
+ buf, numRows, numCols)
-def _copyRowBlock(i, sc, ret, src, numRowsPerBlock, rlen, clen):
+
+def _copyRowBlock(i, sc, ret, src, numRowsPerBlock, rlen, clen):
rowIndex = int(i / numRowsPerBlock)
- tmp = src[i:min(i+numRowsPerBlock, rlen),]
- mb = _convertSPMatrixToMB(sc, tmp) if isinstance(src, spmatrix) else _convertDenseMatrixToMB(sc, tmp)
- sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.copyRowBlocks(mb, rowIndex, ret, numRowsPerBlock, rlen, clen)
+ tmp = src[i:min(i + numRowsPerBlock, rlen), ]
+ mb = _convertSPMatrixToMB(
+ sc,
+ tmp) if isinstance(
+ src,
+ spmatrix) else _convertDenseMatrixToMB(
+ sc,
+ tmp)
+ sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.copyRowBlocks(
+ mb, rowIndex, ret, numRowsPerBlock, rlen, clen)
return i
+
def convertToMatrixBlock(sc, src, maxSizeBlockInMB=8):
if not isinstance(sc, SparkContext):
raise TypeError('sc needs to be of type SparkContext')
@@ -189,33 +250,49 @@ def convertToMatrixBlock(sc, src, maxSizeBlockInMB=8):
src = np.asarray(src, dtype=np.float64) if not isSparse else src
if len(src.shape) != 2:
src_type = str(type(src).__name__)
- raise TypeError('Expected 2-dimensional ' + src_type + ', instead passed ' + str(len(src.shape)) + '-dimensional ' + src_type)
+ raise TypeError('Expected 2-dimensional ' +
+ src_type +
+ ', instead passed ' +
+ str(len(src.shape)) +
+ '-dimensional ' +
+ src_type)
# Ignoring sparsity for computing numRowsPerBlock for now
- numRowsPerBlock = int(math.ceil((maxSizeBlockInMB*1000000) / (src.shape[1]*8)))
+ numRowsPerBlock = int(
+ math.ceil((maxSizeBlockInMB * 1000000) / (src.shape[1] * 8)))
multiBlockTransfer = False if numRowsPerBlock >= src.shape[0] else True
if not multiBlockTransfer:
- return _convertSPMatrixToMB(sc, src) if isSparse else _convertDenseMatrixToMB(sc, src)
+ return _convertSPMatrixToMB(
+ sc, src) if isSparse else _convertDenseMatrixToMB(sc, src)
else:
# Since coo_matrix does not have range indexing
src = csr_matrix(src) if isSparse else src
rlen = int(src.shape[0])
clen = int(src.shape[1])
- ret = sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.allocateDenseOrSparse(rlen, clen, isSparse)
- [ _copyRowBlock(i, sc, ret, src, numRowsPerBlock, rlen, clen) for i in range(0, src.shape[0], numRowsPerBlock) ]
- sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.postProcessAfterCopying(ret)
+ ret = sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.allocateDenseOrSparse(
+ rlen, clen, isSparse)
+ [_copyRowBlock(i, sc, ret, src, numRowsPerBlock, rlen, clen)
+ for i in range(0, src.shape[0], numRowsPerBlock)]
+ sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.postProcessAfterCopying(
+ ret)
return ret
+
def convertToNumPyArr(sc, mb):
if isinstance(sc, SparkContext):
numRows = mb.getNumRows()
numCols = mb.getNumColumns()
createJavaObject(sc, 'dummy')
- buf = sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.convertMBtoPy4JDenseArr(mb)
- return np.frombuffer(buf, count=numRows*numCols, dtype=np.float64).reshape((numRows, numCols))
+ buf = sc._jvm.org.apache.sysml.runtime.instructions.spark.utils.RDDConverterUtilsExt.convertMBtoPy4JDenseArr(
+ mb)
+ return np.frombuffer(buf, count=numRows * numCols,
+ dtype=np.float64).reshape((numRows, numCols))
else:
- raise TypeError('sc needs to be of type SparkContext') # TODO: We can generalize this by creating py4j gateway ourselves
+ # TODO: We can generalize this by creating py4j gateway ourselves
+ raise TypeError('sc needs to be of type SparkContext')
# Returns the mean of a model if defined otherwise None
+
+
def getDatasetMean(dataset_name):
"""
Parameters
@@ -230,7 +307,7 @@ def getDatasetMean(dataset_name):
try:
mean = DATASET_MEAN[dataset_name.upper()]
- except:
+ except BaseException:
mean = None
return mean
@@ -238,14 +315,15 @@ def getDatasetMean(dataset_name):
# Example usage: convertImageToNumPyArr(im, img_shape=(3, 224, 224), add_rotated_images=True, add_mirrored_images=True)
# The above call returns a numpy array of shape (6, 50176) in NCHW format
def convertImageToNumPyArr(im, img_shape=None, add_rotated_images=False, add_mirrored_images=False,
- color_mode = 'RGB', mean=None):
+ color_mode='RGB', mean=None):
- ## Input Parameters
+ # Input Parameters
# color_mode: In case of VGG models which expect image data in BGR format instead of RGB for other most models,
# color_mode parameter is used to process image data in BGR format.
- # mean: mean value is used to subtract from input data from every pixel value. By default value is None, so mean value not subtracted.
+ # mean: mean value is used to subtract from input data from every pixel
+ # value. By default value is None, so mean value not subtracted.
if img_shape is not None:
num_channels = img_shape[0]
@@ -275,7 +353,7 @@ def convertImageToNumPyArr(im, img_shape=None, add_rotated_images=False, add_mir
# RGB -> BGR
if color_mode == 'BGR':
- im = im[...,::-1]
+ im = im[..., ::-1]
# Subtract Mean
if mean is not None:
@@ -288,13 +366,23 @@ def convertImageToNumPyArr(im, img_shape=None, add_rotated_images=False, add_mir
ret = _im2NumPy(im)
if add_rotated_images:
- ret = np.vstack((ret, _im2NumPy(im.rotate(90)), _im2NumPy(im.rotate(180)), _im2NumPy(im.rotate(270)) ))
+ ret = np.vstack(
+ (ret, _im2NumPy(
+ im.rotate(90)), _im2NumPy(
+ im.rotate(180)), _im2NumPy(
+ im.rotate(270))))
if add_mirrored_images:
- ret = np.vstack((ret, _im2NumPy(im.transpose(Image.FLIP_LEFT_RIGHT)), _im2NumPy(im.transpose(Image.FLIP_TOP_BOTTOM))))
+ ret = np.vstack(
+ (ret, _im2NumPy(
+ im.transpose(
+ Image.FLIP_LEFT_RIGHT)), _im2NumPy(
+ im.transpose(
+ Image.FLIP_TOP_BOTTOM))))
return ret
def convertToPandasDF(X):
if not isinstance(X, pd.DataFrame):
- return pd.DataFrame(X, columns=['C' + str(i) for i in range(getNumCols(X))])
+ return pd.DataFrame(X, columns=['C' + str(i)
+ for i in range(getNumCols(X))])
return X
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/defmatrix.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/defmatrix.py b/src/main/python/systemml/defmatrix.py
index c177a05..cbdec4d 100644
--- a/src/main/python/systemml/defmatrix.py
+++ b/src/main/python/systemml/defmatrix.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,9 +17,19 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
-
-__all__ = [ 'setSparkContext', 'matrix', 'eval', 'solve', 'DMLOp', 'set_lazy', 'debug_array_conversion', 'load', 'full', 'seq' ]
+# -------------------------------------------------------------
+
+__all__ = [
+ 'setSparkContext',
+ 'matrix',
+ 'eval',
+ 'solve',
+ 'DMLOp',
+ 'set_lazy',
+ 'debug_array_conversion',
+ 'load',
+ 'full',
+ 'seq']
import numpy as np
import pandas as pd
@@ -31,11 +41,13 @@ try:
from pyspark.sql import DataFrame, SparkSession
import pyspark.mllib.common
except ImportError:
- raise ImportError('Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
+ raise ImportError(
+ 'Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
from . import MLContext, pydml, _java2py, Matrix
from .converters import *
+
def setSparkContext(sc):
"""
Before using the matrix, the user needs to invoke this function if SparkContext is not previously created in the session.
@@ -55,43 +67,48 @@ def check_MLContext():
if SparkContext._active_spark_context is not None:
setSparkContext(SparkContext._active_spark_context)
else:
- raise Exception('Expected setSparkContext(sc) to be called, where sc is active SparkContext.')
+ raise Exception(
+ 'Expected setSparkContext(sc) to be called, where sc is active SparkContext.')
+
+########################## AST related operations ########################
-########################## AST related operations ##################################
class DMLOp(object):
"""
Represents an intermediate node of Abstract syntax tree created to generate the PyDML script
"""
+
def __init__(self, inputs, dml=None):
self.inputs = inputs
self.dml = dml
self.ID = None
self.depth = 1
for m in self.inputs:
- m.referenced = m.referenced + [ self ]
+ m.referenced = m.referenced + [self]
if isinstance(m, matrix) and m.op is not None:
self.depth = max(self.depth, m.op.depth + 1)
MAX_DEPTH = 0
-
+
def _visit(self, execute=True):
matrix.dml = matrix.dml + self.dml
def _print_ast(self, numSpaces):
ret = []
for m in self.inputs:
- ret = [ m._print_ast(numSpaces+2) ]
+ ret = [m._print_ast(numSpaces + 2)]
return ''.join(ret)
+
# Special object used internally to specify the placeholder which will be replaced by output ID
# This helps to provide dml containing output ID in construct_intermediate_node
OUTPUT_ID = '$$OutputID$$'
+
def set_lazy(isLazy):
"""
This method allows users to set whether the matrix operations should be executed in lazy manner.
-
+
Parameters
----------
isLazy: True if matrix operations should be evaluated in lazy manner.
@@ -100,7 +117,8 @@ def set_lazy(isLazy):
DMLOp.MAX_DEPTH = 0
else:
DMLOp.MAX_DEPTH = 1
-
+
+
def construct_intermediate_node(inputs, dml):
"""
Convenient utility to create an intermediate node of AST.
@@ -112,11 +130,12 @@ def construct_intermediate_node(inputs, dml):
"""
dmlOp = DMLOp(inputs)
out = matrix(None, op=dmlOp)
- dmlOp.dml = [out.ID if x==OUTPUT_ID else x for x in dml]
+ dmlOp.dml = [out.ID if x == OUTPUT_ID else x for x in dml]
if DMLOp.MAX_DEPTH > 0 and out.op.depth >= DMLOp.MAX_DEPTH:
out.eval()
return out
+
def load(file, format='csv'):
"""
Allows user to load a matrix from filesystem
@@ -126,7 +145,9 @@ def load(file, format='csv'):
file: filepath
format: can be csv, text or binary or mm
"""
- return construct_intermediate_node([], [OUTPUT_ID, ' = load(\"', file, '\", format=\"', format, '\")\n'])
+ return construct_intermediate_node(
+ [], [OUTPUT_ID, ' = load(\"', file, '\", format=\"', format, '\")\n'])
+
def full(shape, fill_value):
"""
@@ -137,7 +158,9 @@ def full(shape, fill_value):
shape: tuple of length 2
fill_value: float or int
"""
- return construct_intermediate_node([], [OUTPUT_ID, ' = full(', str(fill_value), ', rows=', str(shape[0]), ', cols=', str(shape[1]), ')\n'])
+ return construct_intermediate_node([], [OUTPUT_ID, ' = full(', str(
+ fill_value), ', rows=', str(shape[0]), ', cols=', str(shape[1]), ')\n'])
+
def reset():
"""
@@ -150,6 +173,7 @@ def reset():
matrix.dml = []
matrix.script = pydml('')
+
def perform_dfs(outputs, execute):
"""
Traverses the forest of nodes rooted at outputs nodes and returns the DML script to execute
@@ -167,8 +191,9 @@ def perform_dfs(outputs, execute):
def _log_base(val, base):
if not isinstance(val, str):
raise ValueError('The val to _log_base should be of type string')
- return '(log(' + val + ')/log(' + str(base) + '))'
-
+ return '(log(' + val + ')/log(' + str(base) + '))'
+
+
def _matricize(lhs, inputs):
"""
Utility fn to convert the supported types to matrix class or to string (if float or int)
@@ -184,7 +209,8 @@ def _matricize(lhs, inputs):
else:
raise TypeError('Incorrect type')
return lhsStr, inputs
-
+
+
def binary_op(lhs, rhs, opStr):
"""
Common function called by all the binary operators in matrix class
@@ -192,7 +218,9 @@ def binary_op(lhs, rhs, opStr):
inputs = []
lhsStr, inputs = _matricize(lhs, inputs)
rhsStr, inputs = _matricize(rhs, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', lhsStr, opStr, rhsStr, '\n'])
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', lhsStr, opStr, rhsStr, '\n'])
+
def binaryMatrixFunction(X, Y, fnName):
"""
@@ -201,7 +229,9 @@ def binaryMatrixFunction(X, Y, fnName):
inputs = []
lhsStr, inputs = _matricize(X, inputs)
rhsStr, inputs = _matricize(Y, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', fnName,'(', lhsStr, ', ', rhsStr, ')\n'])
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', fnName, '(', lhsStr, ', ', rhsStr, ')\n'])
+
def unaryMatrixFunction(X, fnName):
"""
@@ -209,14 +239,16 @@ def unaryMatrixFunction(X, fnName):
"""
inputs = []
lhsStr, inputs = _matricize(X, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', fnName,'(', lhsStr, ')\n'])
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', fnName, '(', lhsStr, ')\n'])
+
def seq(start=None, stop=None, step=1):
"""
Creates a single column vector with values starting from <start>, to <stop>, in increments of <step>.
Note: Unlike Numpy's arange which returns a row-vector, this returns a column vector.
Also, Unlike Numpy's arange which doesnot include stop, this method includes stop in the interval.
-
+
Parameters
----------
start: int or float [Optional: default = 0]
@@ -228,9 +260,12 @@ def seq(start=None, stop=None, step=1):
elif start is not None and stop is None:
stop = start
start = 0
- return construct_intermediate_node([], [OUTPUT_ID, ' = seq(', str(start), ',', str(stop), ',', str(step), ')\n'])
-
+ return construct_intermediate_node(
+ [], [OUTPUT_ID, ' = seq(', str(start), ',', str(stop), ',', str(step), ')\n'])
+
# utility function that converts 1:3 into DML string
+
+
def convert_seq_to_dml(s):
ret = []
if s is None:
@@ -239,31 +274,36 @@ def convert_seq_to_dml(s):
if s.step is not None:
raise ValueError('Slicing with step is not supported.')
if s.start is None:
- ret = ret + [ '0 : ' ]
+ ret = ret + ['0 : ']
else:
- ret = ret + [ getValue(s.start), ':' ]
+ ret = ret + [getValue(s.start), ':']
if s.start is None:
- ret = ret + [ '' ]
+ ret = ret + ['']
else:
- ret = ret + [ getValue(s.stop) ]
+ ret = ret + [getValue(s.stop)]
else:
- ret = ret + [ getValue(s) ]
+ ret = ret + [getValue(s)]
return ''.join(ret)
# utility function that converts index (such as [1, 2:3]) into DML string
+
+
def getIndexingDML(index):
- ret = [ '[' ]
+ ret = ['[']
if isinstance(index, tuple) and len(index) == 1:
- ret = ret + [ convert_seq_to_dml(index[0]), ',' ]
+ ret = ret + [convert_seq_to_dml(index[0]), ',']
elif isinstance(index, tuple) and len(index) == 2:
- ret = ret + [ convert_seq_to_dml(index[0]), ',', convert_seq_to_dml(index[1]) ]
+ ret = ret + [convert_seq_to_dml(index[0]),
+ ',', convert_seq_to_dml(index[1])]
else:
- raise TypeError('matrix indexes can only be tuple of length 2. For example: m[1,1], m[0:1,], m[:, 0:1]')
- return ret + [ ']' ]
+ raise TypeError(
+ 'matrix indexes can only be tuple of length 2. For example: m[1,1], m[0:1,], m[:, 0:1]')
+ return ret + [']']
+
def convert_outputs_to_list(outputs):
if isinstance(outputs, matrix):
- return [ outputs ]
+ return [outputs]
elif isinstance(outputs, list):
for o in outputs:
if not isinstance(o, matrix):
@@ -272,10 +312,11 @@ def convert_outputs_to_list(outputs):
else:
raise TypeError('Only matrix or list of matrix allowed')
+
def reset_output_flag(outputs):
for m in outputs:
m.output = False
-
+
###############################################################################
@@ -309,6 +350,7 @@ def solve(A, b):
"""
return binaryMatrixFunction(A, b, 'solve')
+
def eval(outputs, execute=True):
"""
Executes the unevaluated DML script and computes the matrices specified by outputs.
@@ -333,35 +375,37 @@ def eval(outputs, execute=True):
def debug_array_conversion(throwError):
matrix.THROW_ARRAY_CONVERSION_ERROR = throwError
-
+
+
def _get_new_var_id():
matrix.systemmlVarID += 1
return 'mVar' + str(matrix.systemmlVarID)
###############################################################################
+
class matrix(object):
"""
matrix class is a python wrapper that implements basic matrix operators, matrix functions
as well as converters to common Python types (for example: Numpy arrays, PySpark DataFrame
- and Pandas DataFrame).
-
+ and Pandas DataFrame).
+
The operators supported are:
-
+
1. Arithmetic operators: +, -, *, /, //, %, ** as well as dot (i.e. matrix multiplication)
2. Indexing in the matrix
3. Relational/Boolean operators: <, <=, >, >=, ==, !=, &, |
-
+
In addition, following functions are supported for matrix:
-
+
1. transpose
2. Aggregation functions: sum, mean, var, sd, max, min, argmin, argmax, cumsum
3. Global statistical built-In functions: exp, log, abs, sqrt, round, floor, ceil, ceiling, sin, cos, tan, asin, acos, atan, sign, solve
-
- For all the above functions, we always return a two dimensional matrix, especially for aggregation functions with axis.
+
+ For all the above functions, we always return a two dimensional matrix, especially for aggregation functions with axis.
For example: Assuming m1 is a matrix of (3, n), NumPy returns a 1d vector of dimension (3,) for operation m1.sum(axis=1)
whereas SystemML returns a 2d matrix of dimension (3, 1).
-
+
Note: an evaluated matrix contains a data field computed by eval method as DataFrame or NumPy array.
Examples
@@ -396,9 +440,9 @@ class matrix(object):
array([[-60.],
[-60.],
[-60.]])
-
+
Design Decisions:
-
+
1. Until eval() method is invoked, we create an AST (not exposed to the user) that consist of unevaluated operations and data required by those operations.
As an anology, a spark user can treat eval() method similar to calling RDD.persist() followed by RDD.count().
2. The AST consist of two kinds of nodes: either of type matrix or of type DMLOp.
@@ -408,7 +452,7 @@ class matrix(object):
If not evaluated, the attribute 'op' which refers to one of the intermediate node of AST and if of type DMLOp. In this case, the attribute 'data' is set to None.
4. DMLOp has an attribute 'inputs' which contains list of matrix objects or DMLOp.
5. To simplify the traversal, every matrix object is considered immutable and an matrix operations creates a new matrix object.
- As an example:
+ As an example:
`m1 = sml.matrix(np.ones((3,3)))` creates a matrix object backed by 'data=(np.ones((3,3))'.
`m1 = m1 * 2` will create a new matrix object which is now backed by 'op=DMLOp( ... )' whose input is earlier created matrix object.
6. Left indexing (implemented in __setitem__ method) is a special case, where Python expects the existing object to be mutated.
@@ -417,7 +461,7 @@ class matrix(object):
left-indexed-matrix = new-deep-copied-matrix
left-indexed-matrix[index] = value
7. Please use m.print_ast() and/or type `m` for debugging. Here is a sample session:
-
+
>>> npm = np.ones((3,3))
>>> m1 = sml.matrix(npm + 3)
>>> m2 = sml.matrix(npm + 5)
@@ -430,14 +474,16 @@ class matrix(object):
>>> m3.print_ast()
- [mVar3] (op).
- [mVar1] (data).
- - [mVar2] (data).
-
+ - [mVar2] (data).
+
"""
- # Global variable that is used to keep track of intermediate matrix variables in the DML script
+ # Global variable that is used to keep track of intermediate matrix
+ # variables in the DML script
systemmlVarID = 0
# Since joining of string is expensive operation, we collect the set of strings into list and then join
- # them before execution: See matrix.script.scriptString = ''.join(matrix.dml) in eval() method
+ # them before execution: See matrix.script.scriptString =
+ # ''.join(matrix.dml) in eval() method
dml = []
# Represents MLContext's script object
@@ -449,7 +495,7 @@ class matrix(object):
# Contains list of nodes visited in Abstract Syntax Tree. This helps to avoid computation of matrix objects
# that have been previously evaluated.
visited = []
-
+
def __init__(self, data, op=None):
"""
Constructs a lazy matrix
@@ -464,13 +510,15 @@ class matrix(object):
self.output = False
self.ID = _get_new_var_id()
self.referenced = []
- # op refers to the node of Abstract Syntax Tree created internally for lazy evaluation
+ # op refers to the node of Abstract Syntax Tree created internally for
+ # lazy evaluation
self.op = op
self.eval_data = data
self._shape = None
if isinstance(data, SUPPORTED_TYPES):
self._shape = data.shape
- if not (isinstance(data, SUPPORTED_TYPES) or hasattr(data, '_jdf') or (data is None and op is not None)):
+ if not (isinstance(data, SUPPORTED_TYPES) or hasattr(
+ data, '_jdf') or (data is None and op is not None)):
raise TypeError('Unsupported input type')
def eval(self):
@@ -478,14 +526,15 @@ class matrix(object):
This is a convenience function that calls the global eval method
"""
eval([self])
-
+
def toPandas(self):
"""
This is a convenience function that calls the global eval method and then converts the matrix object into Pandas DataFrame.
"""
self.eval()
if isinstance(self.eval_data, py4j.java_gateway.JavaObject):
- self.eval_data = _java2py(SparkContext._active_spark_context, self.eval_data)
+ self.eval_data = _java2py(
+ SparkContext._active_spark_context, self.eval_data)
if isinstance(self.eval_data, Matrix):
self.eval_data = self.eval_data.toNumPy()
self.eval_data = convertToPandasDF(self.eval_data)
@@ -497,7 +546,8 @@ class matrix(object):
"""
self.eval()
if isinstance(self.eval_data, py4j.java_gateway.JavaObject):
- self.eval_data = _java2py(SparkContext._active_spark_context, self.eval_data)
+ self.eval_data = _java2py(
+ SparkContext._active_spark_context, self.eval_data)
if isinstance(self.eval_data, Matrix):
self.eval_data = self.eval_data.toNumPy()
return self.eval_data
@@ -519,7 +569,8 @@ class matrix(object):
if isinstance(self.eval_data, DataFrame):
return self.eval_data
if isinstance(self.eval_data, py4j.java_gateway.JavaObject):
- self.eval_data = _java2py(SparkContext._active_spark_context, self.eval_data)
+ self.eval_data = _java2py(
+ SparkContext._active_spark_context, self.eval_data)
if isinstance(self.eval_data, Matrix):
self.eval_data = self.eval_data.toDF()
return self.eval_data
@@ -529,36 +580,40 @@ class matrix(object):
def save(self, file, format='csv'):
"""
Allows user to save a matrix to filesystem
-
+
Parameters
----------
file: filepath
format: can be csv, text or binary or mm
"""
- tmp = construct_intermediate_node([self], ['save(', self.ID , ',\"', file, '\", format=\"', format, '\")\n'])
- construct_intermediate_node([tmp], [OUTPUT_ID, ' = full(0, rows=1, cols=1)\n']).eval()
-
+ tmp = construct_intermediate_node(
+ [self], ['save(', self.ID, ',\"', file, '\", format=\"', format, '\")\n'])
+ construct_intermediate_node(
+ [tmp], [OUTPUT_ID, ' = full(0, rows=1, cols=1)\n']).eval()
+
def _mark_as_visited(self):
self.visited = True
# for cleanup
- matrix.visited = matrix.visited + [ self ]
+ matrix.visited = matrix.visited + [self]
return self
def _register_as_input(self, execute):
# TODO: Remove this when automatic registration of frame is resolved
- matrix.dml = [ self.ID, ' = load(\" \", format=\"csv\")\n'] + matrix.dml
+ matrix.dml = [self.ID, ' = load(\" \", format=\"csv\")\n'] + matrix.dml
if isinstance(self.eval_data, SUPPORTED_TYPES) and execute:
- matrix.script.input(self.ID, convertToMatrixBlock(matrix.sc, self.eval_data))
+ matrix.script.input(
+ self.ID, convertToMatrixBlock(
+ matrix.sc, self.eval_data))
elif execute:
matrix.script.input(self.ID, self.toDF())
return self
def _register_as_output(self, execute):
# TODO: Remove this when automatic registration of frame is resolved
- matrix.dml = matrix.dml + ['save(', self.ID, ', \" \")\n']
+ matrix.dml = matrix.dml + ['save(', self.ID, ', \" \")\n']
if execute:
matrix.script.output(self.ID)
-
+
def _visit(self, execute=True):
"""
This function is called for two scenarios:
@@ -585,7 +640,7 @@ class matrix(object):
def print_ast(self):
"""
Please use m.print_ast() and/or type `m` for debugging. Here is a sample session:
-
+
>>> npm = np.ones((3,3))
>>> m1 = sml.matrix(npm + 3)
>>> m2 = sml.matrix(npm + 5)
@@ -601,15 +656,15 @@ class matrix(object):
- [mVar2] (data).
"""
return self._print_ast(0)
-
+
def _print_ast(self, numSpaces):
- head = ''.join([ ' ' ]*numSpaces + [ '- [', self.ID, '] ' ])
+ head = ''.join([' '] * numSpaces + ['- [', self.ID, '] '])
if self.eval_data is not None:
out = head + '(data).\n'
elif self.op is not None:
- ret = [ head, '(op).\n' ]
+ ret = [head, '(op).\n']
for m in self.op.inputs:
- ret = ret + [ m._print_ast(numSpaces + 2) ]
+ ret = ret + [m._print_ast(numSpaces + 2)]
out = ''.join(ret)
else:
raise ValueError('Either op or data needs to be set')
@@ -623,66 +678,81 @@ class matrix(object):
This function helps to debug matrix class and also examine the generated PyDML script
"""
if self.eval_data is None:
- print('# This matrix (' + self.ID + ') is backed by below given PyDML script (which is not yet evaluated). To fetch the data of this matrix, invoke toNumPy() or toDF() or toPandas() methods.\n' + eval([self], execute=False))
+ print(
+ '# This matrix (' +
+ self.ID +
+ ') is backed by below given PyDML script (which is not yet evaluated). To fetch the data of this matrix, invoke toNumPy() or toDF() or toPandas() methods.\n' +
+ eval(
+ [self],
+ execute=False))
else:
- print('# This matrix (' + self.ID + ') is backed by ' + str(type(self.eval_data)) + '. To fetch the DataFrame or NumPy array, invoke toDF() or toNumPy() method respectively.')
+ print('# This matrix (' + self.ID + ') is backed by ' + str(type(self.eval_data)) +
+ '. To fetch the DataFrame or NumPy array, invoke toDF() or toNumPy() method respectively.')
return ''
-
- ######################### NumPy related methods ######################################
-
+
+ ######################### NumPy related methods ##########################
+
__array_priority__ = 10.2
ndim = 2
-
+
THROW_ARRAY_CONVERSION_ERROR = False
-
+
def __array__(self, dtype=np.double):
"""
As per NumPy from Python,
This method is called to obtain an ndarray object when needed. You should always guarantee this returns an actual ndarray object.
-
+
Using this method, you get back a ndarray object, and subsequent operations on the returned ndarray object will be singlenode.
"""
if not isinstance(self.eval_data, SUPPORTED_TYPES):
- # Only warn if there is an unevaluated operation (which could potentially generate large matrix or if data is non-supported singlenode formats)
+ # Only warn if there is an unevaluated operation (which could
+ # potentially generate large matrix or if data is non-supported
+ # singlenode formats)
import inspect
- frame,filename,line_number,function_name,lines,index = inspect.stack()[1]
- msg = 'Conversion from SystemML matrix to NumPy array (occurs in ' + str(filename) + ':' + str(line_number) + ' ' + function_name + ")"
+ frame, filename, line_number, function_name, lines, index = inspect.stack()[
+ 1]
+ msg = 'Conversion from SystemML matrix to NumPy array (occurs in ' + str(
+ filename) + ':' + str(line_number) + ' ' + function_name + ")"
if matrix.THROW_ARRAY_CONVERSION_ERROR:
raise Exception('[ERROR]:' + msg)
else:
print('[WARN]:' + msg)
return np.array(self.toNumPy(), dtype)
-
+
def astype(self, t):
# TODO: Throw error if incorrect type
return self
-
+
def asfptype(self):
return self
-
- def set_shape(self,shape):
+
+ def set_shape(self, shape):
raise NotImplementedError('Reshaping is not implemented')
-
+
def get_shape(self):
if self._shape is None:
lhsStr, inputs = _matricize(self, [])
rlen_ID = _get_new_var_id()
clen_ID = _get_new_var_id()
multiline_dml = [rlen_ID, ' = ', lhsStr, '.shape(0)\n']
- multiline_dml = multiline_dml + [clen_ID, ' = ', lhsStr, '.shape(1)\n']
- multiline_dml = multiline_dml + [OUTPUT_ID, ' = full(0, rows=2, cols=1)\n']
- multiline_dml = multiline_dml + [ OUTPUT_ID, '[0,0] = ', rlen_ID, '\n' ]
- multiline_dml = multiline_dml + [ OUTPUT_ID, '[1,0] = ', clen_ID, '\n' ]
+ multiline_dml = multiline_dml + \
+ [clen_ID, ' = ', lhsStr, '.shape(1)\n']
+ multiline_dml = multiline_dml + \
+ [OUTPUT_ID, ' = full(0, rows=2, cols=1)\n']
+ multiline_dml = multiline_dml + \
+ [OUTPUT_ID, '[0,0] = ', rlen_ID, '\n']
+ multiline_dml = multiline_dml + \
+ [OUTPUT_ID, '[1,0] = ', clen_ID, '\n']
ret = construct_intermediate_node(inputs, multiline_dml).toNumPy()
self._shape = tuple(np.array(ret, dtype=int).flatten())
- return self._shape
-
+ return self._shape
+
shape = property(fget=get_shape, fset=set_shape)
-
+
def __numpy_ufunc__(self, func, method, pos, inputs, **kwargs):
"""
This function enables systemml matrix to be compatible with NumPy's ufuncs.
-
+
Parameters
----------
func: ufunc object that was called.
@@ -699,7 +769,7 @@ class matrix(object):
return NotImplemented
if len(inputs) == 2:
return fn(inputs[0], inputs[1])
- elif len(inputs) == 1:
+ elif len(inputs) == 1:
return fn(inputs[0])
else:
raise ValueError('Unsupported number of inputs')
@@ -709,98 +779,111 @@ class matrix(object):
Stack matrices horizontally (column wise). Invokes cbind internally.
"""
return binaryMatrixFunction(self, other, 'cbind')
-
+
def vstack(self, other):
"""
Stack matrices vertically (row wise). Invokes rbind internally.
"""
return binaryMatrixFunction(self, other, 'rbind')
-
- ######################### Arithmetic operators ######################################
+
+ ######################### Arithmetic operators ###########################
def negative(self):
lhsStr, inputs = _matricize(self, [])
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = -', lhsStr, '\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = -', lhsStr, '\n'])
+
def remainder(self, other):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rhsStr, inputs = _matricize(other, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = floor(', lhsStr, '/', rhsStr, ') * ', rhsStr, '\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = floor(', lhsStr, '/', rhsStr, ') * ', rhsStr, '\n'])
+
def ldexp(self, other):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rhsStr, inputs = _matricize(other, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', lhsStr, '* (2**', rhsStr, ')\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', lhsStr, '* (2**', rhsStr, ')\n'])
+
def mod(self, other):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rhsStr, inputs = _matricize(other, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', lhsStr, ' - floor(', lhsStr, '/', rhsStr, ') * ', rhsStr, '\n'])
-
+ return construct_intermediate_node(inputs, [
+ OUTPUT_ID, ' = ', lhsStr, ' - floor(', lhsStr, '/', rhsStr, ') * ', rhsStr, '\n'])
+
def logaddexp(self, other):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rhsStr, inputs = _matricize(other, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = log(exp(', lhsStr, ') + exp(', rhsStr, '))\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = log(exp(', lhsStr, ') + exp(', rhsStr, '))\n'])
+
def logaddexp2(self, other):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rhsStr, inputs = _matricize(other, inputs)
- opStr = _log_base('2**' + lhsStr + '2**' + rhsStr, 2)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', opStr, '\n'])
+ opStr = _log_base('2**' + lhsStr + '2**' + rhsStr, 2)
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', opStr, '\n'])
def log1p(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = log(1 + ', lhsStr, ')\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = log(1 + ', lhsStr, ')\n'])
+
def exp(self):
return unaryMatrixFunction(self, 'exp')
def exp2(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = 2**', lhsStr, '\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = 2**', lhsStr, '\n'])
+
def square(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', lhsStr, '**2\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', lhsStr, '**2\n'])
+
def reciprocal(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = 1/', lhsStr, '\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = 1/', lhsStr, '\n'])
+
def expm1(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = exp(', lhsStr, ') - 1\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = exp(', lhsStr, ') - 1\n'])
+
def ones_like(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rlen = lhsStr + '.shape(axis=0)'
clen = lhsStr + '.shape(axis=1)'
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = full(1, rows=', rlen, ', cols=', clen, ')\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = full(1, rows=', rlen, ', cols=', clen, ')\n'])
+
def zeros_like(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
rlen = lhsStr + '.shape(axis=0)'
clen = lhsStr + '.shape(axis=1)'
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = full(0, rows=', rlen, ', cols=', clen, ')\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = full(0, rows=', rlen, ', cols=', clen, ')\n'])
+
def log2(self):
return self.log(2)
-
+
def log10(self):
return self.log(10)
-
+
def log(self, y=None):
if y is None:
return unaryMatrixFunction(self, 'log')
@@ -833,7 +916,7 @@ class matrix(object):
def tan(self):
return unaryMatrixFunction(self, 'tan')
-
+
def sinh(self):
return unaryMatrixFunction(self, 'sinh')
@@ -842,7 +925,7 @@ class matrix(object):
def tanh(self):
return unaryMatrixFunction(self, 'tanh')
-
+
def arcsin(self):
return self.asin()
@@ -851,7 +934,7 @@ class matrix(object):
def arctan(self):
return self.atan()
-
+
def asin(self):
return unaryMatrixFunction(self, 'asin')
@@ -868,8 +951,9 @@ class matrix(object):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
# 180/pi = 57.2957795131
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', lhsStr, '*57.2957795131\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', lhsStr, '*57.2957795131\n'])
+
def deg2rad(self):
"""
Convert angles from degrees to radians.
@@ -877,10 +961,11 @@ class matrix(object):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
# pi/180 = 0.01745329251
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = ', lhsStr, '*0.01745329251\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = ', lhsStr, '*0.01745329251\n'])
+
def sign(self):
- return unaryMatrixFunction(self, 'sign')
+ return unaryMatrixFunction(self, 'sign')
def __add__(self, other):
return binary_op(self, other, ' + ')
@@ -951,8 +1036,7 @@ class matrix(object):
"""
return binaryMatrixFunction(self, other, 'dot')
-
- ######################### Relational/Boolean operators ######################################
+ ######################### Relational/Boolean operators ###################
def __lt__(self, other):
return binary_op(self, other, ' < ')
@@ -964,14 +1048,14 @@ class matrix(object):
return binary_op(self, other, ' > ')
def __ge__(self, other):
- return binary_op(self, other,' >= ')
+ return binary_op(self, other, ' >= ')
def __eq__(self, other):
return binary_op(self, other, ' == ')
def __ne__(self, other):
return binary_op(self, other, ' != ')
-
+
# TODO: Cast the output back into scalar and return boolean results
def __and__(self, other):
return binary_op(other, self, ' & ')
@@ -982,12 +1066,13 @@ class matrix(object):
def logical_not(self):
inputs = []
lhsStr, inputs = _matricize(self, inputs)
- return construct_intermediate_node(inputs, [OUTPUT_ID, ' = !', lhsStr, '\n'])
-
+ return construct_intermediate_node(
+ inputs, [OUTPUT_ID, ' = !', lhsStr, '\n'])
+
def remove_empty(self, axis=None):
"""
Removes all empty rows or columns from the input matrix target X according to specified axis.
-
+
Parameters
----------
axis : int (0 or 1)
@@ -995,16 +1080,19 @@ class matrix(object):
if axis is None:
raise ValueError('axis is a mandatory argument for remove_empty')
if axis == 0:
- return self._parameterized_helper_fn(self, 'removeEmpty', { 'target':self, 'margin':'rows' })
+ return self._parameterized_helper_fn(
+ self, 'removeEmpty', {'target': self, 'margin': 'rows'})
elif axis == 1:
- return self._parameterized_helper_fn(self, 'removeEmpty', { 'target':self, 'margin':'cols' })
+ return self._parameterized_helper_fn(
+ self, 'removeEmpty', {'target': self, 'margin': 'cols'})
else:
- raise ValueError('axis for remove_empty needs to be either 0 or 1.')
-
+ raise ValueError(
+ 'axis for remove_empty needs to be either 0 or 1.')
+
def replace(self, pattern=None, replacement=None):
"""
Removes all empty rows or columns from the input matrix target X according to specified axis.
-
+
Parameters
----------
pattern : float or int
@@ -1014,43 +1102,44 @@ class matrix(object):
raise ValueError('pattern should be of type float or int')
if replacement is None or not isinstance(replacement, (float, int)):
raise ValueError('replacement should be of type float or int')
- return self._parameterized_helper_fn(self, 'replace', { 'target':self, 'pattern':pattern, 'replacement':replacement })
-
+ return self._parameterized_helper_fn(
+ self, 'replace', {'target': self, 'pattern': pattern, 'replacement': replacement})
+
def _parameterized_helper_fn(self, fnName, **kwargs):
"""
Helper to invoke parameterized builtin function
"""
dml_script = ''
lhsStr, inputs = _matricize(self, [])
- dml_script = [OUTPUT_ID, ' = ', fnName, '(', lhsStr ]
+ dml_script = [OUTPUT_ID, ' = ', fnName, '(', lhsStr]
first_arg = True
for key in kwargs:
if first_arg:
first_arg = False
else:
- dml_script = dml_script + [ ', ' ]
+ dml_script = dml_script + [', ']
v = kwargs[key]
if isinstance(v, str):
- dml_script = dml_script + [key, '=\"', v, '\"' ]
+ dml_script = dml_script + [key, '=\"', v, '\"']
elif isinstance(v, matrix):
dml_script = dml_script + [key, '=', v.ID]
else:
- dml_script = dml_script + [key, '=', str(v) ]
- dml_script = dml_script + [ ')\n' ]
+ dml_script = dml_script + [key, '=', str(v)]
+ dml_script = dml_script + [')\n']
return construct_intermediate_node(inputs, dml_script)
-
- ######################### Aggregation functions ######################################
+
+ ######################### Aggregation functions ##########################
def prod(self):
"""
Return the product of all cells in matrix
"""
return self._aggFn('prod', None)
-
+
def sum(self, axis=None):
"""
Compute the sum along the specified axis
-
+
Parameters
----------
axis : int, optional
@@ -1060,7 +1149,7 @@ class matrix(object):
def mean(self, axis=None):
"""
Compute the arithmetic mean along the specified axis
-
+
Parameters
----------
axis : int, optional
@@ -1071,17 +1160,17 @@ class matrix(object):
"""
Compute the variance along the specified axis.
We assume that delta degree of freedom is 1 (unlike NumPy which assumes ddof=0).
-
+
Parameters
----------
axis : int, optional
"""
return self._aggFn('var', axis)
-
+
def moment(self, moment=1, axis=None):
"""
Calculates the nth moment about the mean
-
+
Parameters
----------
moment : int
@@ -1095,31 +1184,60 @@ class matrix(object):
elif moment == 3 or moment == 4:
return self._moment_helper(moment, axis)
else:
- raise ValueError('The specified moment is not supported:' + str(moment))
-
+ raise ValueError(
+ 'The specified moment is not supported:' +
+ str(moment))
+
def _moment_helper(self, k, axis=0):
dml_script = ''
lhsStr, inputs = _matricize(self, [])
- dml_script = [OUTPUT_ID, ' = moment(', lhsStr, ', ', str(k), ')\n' ]
- dml_script = [OUTPUT_ID, ' = moment(', lhsStr, ', ', str(k), ')\n' ]
+ dml_script = [OUTPUT_ID, ' = moment(', lhsStr, ', ', str(k), ')\n']
+ dml_script = [OUTPUT_ID, ' = moment(', lhsStr, ', ', str(k), ')\n']
if axis is None:
- dml_script = [OUTPUT_ID, ' = moment(full(', lhsStr, ', rows=length(', lhsStr, '), cols=1), ', str(k), ')\n' ]
+ dml_script = [
+ OUTPUT_ID,
+ ' = moment(full(',
+ lhsStr,
+ ', rows=length(',
+ lhsStr,
+ '), cols=1), ',
+ str(k),
+ ')\n']
elif axis == 0:
- dml_script = [OUTPUT_ID, ' = full(0, rows=nrow(', lhsStr, '), cols=1)\n' ]
- dml_script = dml_script + [ 'parfor(i in 1:nrow(', lhsStr, '), check=0):\n' ]
- dml_script = dml_script + [ '\t', OUTPUT_ID, '[i-1, 0] = moment(full(', lhsStr, '[i-1,], rows=ncol(', lhsStr, '), cols=1), ', str(k), ')\n\n' ]
+ dml_script = [
+ OUTPUT_ID, ' = full(0, rows=nrow(', lhsStr, '), cols=1)\n']
+ dml_script = dml_script + \
+ ['parfor(i in 1:nrow(', lhsStr, '), check=0):\n']
+ dml_script = dml_script + ['\t',
+ OUTPUT_ID,
+ '[i-1, 0] = moment(full(',
+ lhsStr,
+ '[i-1,], rows=ncol(',
+ lhsStr,
+ '), cols=1), ',
+ str(k),
+ ')\n\n']
elif axis == 1:
- dml_script = [OUTPUT_ID, ' = full(0, rows=1, cols=ncol(', lhsStr, '))\n' ]
- dml_script = dml_script + [ 'parfor(i in 1:ncol(', lhsStr, '), check=0):\n' ]
- dml_script = dml_script + [ '\t', OUTPUT_ID, '[0, i-1] = moment(', lhsStr, '[,i-1], ', str(k), ')\n\n' ]
+ dml_script = [
+ OUTPUT_ID, ' = full(0, rows=1, cols=ncol(', lhsStr, '))\n']
+ dml_script = dml_script + \
+ ['parfor(i in 1:ncol(', lhsStr, '), check=0):\n']
+ dml_script = dml_script + \
+ ['\t',
+ OUTPUT_ID,
+ '[0, i-1] = moment(',
+ lhsStr,
+ '[,i-1], ',
+ str(k),
+ ')\n\n']
else:
raise ValueError('Incorrect axis:' + axis)
return construct_intermediate_node(inputs, dml_script)
-
+
def sd(self, axis=None):
"""
Compute the standard deviation along the specified axis
-
+
Parameters
----------
axis : int, optional
@@ -1129,7 +1247,7 @@ class matrix(object):
def max(self, other=None, axis=None):
"""
Compute the maximum value along the specified axis
-
+
Parameters
----------
other: matrix or numpy array (& other supported types) or scalar
@@ -1145,7 +1263,7 @@ class matrix(object):
def min(self, other=None, axis=None):
"""
Compute the minimum value along the specified axis
-
+
Parameters
----------
other: matrix or numpy array (& other supported types) or scalar
@@ -1161,7 +1279,7 @@ class matrix(object):
def argmin(self, axis=None):
"""
Returns the indices of the minimum values along an axis.
-
+
Parameters
----------
axis : int, optional (only axis=1, i.e. rowIndexMax is supported in this version)
@@ -1171,7 +1289,7 @@ class matrix(object):
def argmax(self, axis=None):
"""
Returns the indices of the maximum values along an axis.
-
+
Parameters
----------
axis : int, optional (only axis=1, i.e. rowIndexMax is supported in this version)
@@ -1181,7 +1299,7 @@ class matrix(object):
def cumsum(self, axis=None):
"""
Returns the indices of the maximum values along an axis.
-
+
Parameters
----------
axis : int, optional (only axis=0, i.e. cumsum along the rows is supported in this version)
@@ -1209,26 +1327,29 @@ class matrix(object):
if axis is None:
dml_script = [OUTPUT_ID, ' = ', fnName, '(', lhsStr, ')\n']
else:
- dml_script = [OUTPUT_ID, ' = ', fnName, '(', lhsStr, ', axis=', str(axis) ,')\n']
+ dml_script = [OUTPUT_ID, ' = ', fnName,
+ '(', lhsStr, ', axis=', str(axis), ')\n']
return construct_intermediate_node(inputs, dml_script)
- ######################### Indexing operators ######################################
+ ######################### Indexing operators #############################
def __getitem__(self, index):
"""
Implements evaluation of right indexing operations such as m[1,1], m[0:1,], m[:, 0:1]
"""
- return construct_intermediate_node([self], [OUTPUT_ID, ' = ', self.ID ] + getIndexingDML(index) + [ '\n' ])
+ return construct_intermediate_node(
+ [self], [OUTPUT_ID, ' = ', self.ID] + getIndexingDML(index) + ['\n'])
# Performs deep copy if the matrix is backed by data
def _prepareForInPlaceUpdate(self):
temp = matrix(self.eval_data, op=self.op)
for op in self.referenced:
- op.inputs = [temp if x.ID==self.ID else x for x in op.inputs]
- self.ID, temp.ID = temp.ID, self.ID # Copy even the IDs as the IDs might be used to create DML
+ op.inputs = [temp if x.ID == self.ID else x for x in op.inputs]
+ # Copy even the IDs as the IDs might be used to create DML
+ self.ID, temp.ID = temp.ID, self.ID
self.op = DMLOp([temp], dml=[self.ID, " = ", temp.ID])
self.eval_data = None
- temp.referenced = self.referenced + [ self.op ]
+ temp.referenced = self.referenced + [self.op]
self.referenced = []
def __setitem__(self, index, value):
@@ -1237,10 +1358,64 @@ class matrix(object):
"""
self._prepareForInPlaceUpdate()
if isinstance(value, matrix) or isinstance(value, DMLOp):
- self.op.inputs = self.op.inputs + [ value ]
+ self.op.inputs = self.op.inputs + [value]
if isinstance(value, matrix):
- value.referenced = value.referenced + [ self.op ]
- self.op.dml = self.op.dml + [ '\n', self.ID ] + getIndexingDML(index) + [ ' = ', getValue(value), '\n']
-
- # Not implemented: conj, hyperbolic/inverse-hyperbolic functions(i.e. sinh, arcsinh, cosh, ...), bitwise operator, xor operator, isreal, iscomplex, isfinite, isinf, isnan, copysign, nextafter, modf, frexp, trunc
- _numpy_to_systeml_mapping = {np.add: __add__, np.subtract: __sub__, np.multiply: __mul__, np.divide: __div__, np.logaddexp: logaddexp, np.true_divide: __truediv__, np.floor_divide: __floordiv__, np.negative: negative, np.power: __pow__, np.remainder: remainder, np.mod: mod, np.fmod: __mod__, np.absolute: abs, np.rint: round, np.sign: sign, np.exp: exp, np.exp2: exp2, np.log: log, np.log2: log2, np.log10: log10, np.expm1: expm1, np.log1p: log1p, np.sqrt: sqrt, np.square: square, np.reciprocal: reciprocal, np.ones_like: ones_like, np.zeros_like: zeros_like, np.sin: sin, np.cos: cos, np.tan: tan, np.arcsin: arcsin, np.arccos: arccos, np.arctan: arctan, np.deg2rad: deg2rad, np.rad2deg: rad2deg, np.greater: __gt__, np.greater_equal: __ge__, np.less: __lt__, np.less_equal: __le__, np.not_equal: __ne__, np.equal: __eq__, np.logical_not: logical_not, np.logical_and: __and__, np.logical_or: __or__, np.maximum: max, np.minimum: min, np.signbit: sign, np.ldexp: ldexp, np.dot:dot}
+ value.referenced = value.referenced + [self.op]
+ self.op.dml = self.op.dml + \
+ ['\n', self.ID] + \
+ getIndexingDML(index) + [' = ', getValue(value), '\n']
+
+ # Not implemented: conj, hyperbolic/inverse-hyperbolic functions(i.e.
+ # sinh, arcsinh, cosh, ...), bitwise operator, xor operator, isreal,
+ # iscomplex, isfinite, isinf, isnan, copysign, nextafter, modf, frexp,
+ # trunc
+ _numpy_to_systeml_mapping = {
+ np.add: __add__,
+ np.subtract: __sub__,
+ np.multiply: __mul__,
+ np.divide: __div__,
+ np.logaddexp: logaddexp,
+ np.true_divide: __truediv__,
+ np.floor_divide: __floordiv__,
+ np.negative: negative,
+ np.power: __pow__,
+ np.remainder: remainder,
+ np.mod: mod,
+ np.fmod: __mod__,
+ np.absolute: abs,
+ np.rint: round,
+ np.sign: sign,
+ np.exp: exp,
+ np.exp2: exp2,
+ np.log: log,
+ np.log2: log2,
+ np.log10: log10,
+ np.expm1: expm1,
+ np.log1p: log1p,
+ np.sqrt: sqrt,
+ np.square: square,
+ np.reciprocal: reciprocal,
+ np.ones_like: ones_like,
+ np.zeros_like: zeros_like,
+ np.sin: sin,
+ np.cos: cos,
+ np.tan: tan,
+ np.arcsin: arcsin,
+ np.arccos: arccos,
+ np.arctan: arctan,
+ np.deg2rad: deg2rad,
+ np.rad2deg: rad2deg,
+ np.greater: __gt__,
+ np.greater_equal: __ge__,
+ np.less: __lt__,
+ np.less_equal: __le__,
+ np.not_equal: __ne__,
+ np.equal: __eq__,
+ np.logical_not: logical_not,
+ np.logical_and: __and__,
+ np.logical_or: __or__,
+ np.maximum: max,
+ np.minimum: min,
+ np.signbit: sign,
+ np.ldexp: ldexp,
+ np.dot: dot}
[2/3] systemml git commit: [MINOR] Refactored the Python code to
match PEP8 guidelines and bugfix in Keras2DML for Python 3.5
Posted by ni...@apache.org.
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/mlcontext.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/mlcontext.py b/src/main/python/systemml/mlcontext.py
index c945b2b..9956815 100644
--- a/src/main/python/systemml/mlcontext.py
+++ b/src/main/python/systemml/mlcontext.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,19 +17,29 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
# Methods to create Script object
-script_factory_methods = [ 'dml', 'pydml', 'dmlFromResource', 'pydmlFromResource', 'dmlFromFile', 'pydmlFromFile', 'dmlFromUrl', 'pydmlFromUrl' ]
+script_factory_methods = [
+ 'dml',
+ 'pydml',
+ 'dmlFromResource',
+ 'pydmlFromResource',
+ 'dmlFromFile',
+ 'pydmlFromFile',
+ 'dmlFromUrl',
+ 'pydmlFromUrl']
# Utility methods
-util_methods = [ '_java2py', 'getHopDAG' ]
-__all__ = ['MLResults', 'MLContext', 'Script', 'Matrix' ] + script_factory_methods + util_methods
+util_methods = ['_java2py', 'getHopDAG']
+__all__ = ['MLResults', 'MLContext', 'Script', 'Matrix'] + \
+ script_factory_methods + util_methods
import os
import numpy as np
import pandas as pd
-import threading, time
-
+import threading
+import time
+
try:
import py4j.java_gateway
from py4j.java_gateway import JavaObject
@@ -38,12 +48,15 @@ try:
import pyspark.mllib.common
from pyspark.sql import SparkSession
except ImportError:
- raise ImportError('Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
+ raise ImportError(
+ 'Unable to import `pyspark`. Hint: Make sure you are running with PySpark.')
from .converters import *
from .classloader import *
-def getHopDAG(ml, script, lines=None, conf=None, apply_rewrites=True, with_subgraph=False):
+
+def getHopDAG(ml, script, lines=None, conf=None,
+ apply_rewrites=True, with_subgraph=False):
"""
Compile a DML / PyDML script.
@@ -51,39 +64,42 @@ def getHopDAG(ml, script, lines=None, conf=None, apply_rewrites=True, with_subgr
----------
ml: MLContext instance
MLContext instance.
-
+
script: Script instance
Script instance defined with the appropriate input and output variables.
-
+
lines: list of integers
Optional: only display the hops that have begin and end line number equals to the given integers.
-
+
conf: SparkConf instance
Optional spark configuration
-
+
apply_rewrites: boolean
If True, perform static rewrites, perform intra-/inter-procedural analysis to propagate size information into functions and apply dynamic rewrites
-
+
with_subgraph: boolean
- If False, the dot graph will be created without subgraphs for statement blocks.
-
+ If False, the dot graph will be created without subgraphs for statement blocks.
+
Returns
-------
hopDAG: string
- hop DAG in dot format
+ hop DAG in dot format
"""
if not isinstance(script, Script):
raise ValueError("Expected script to be an instance of Script")
scriptString = script.scriptString
script_java = script.script_java
- lines = [ int(x) for x in lines ] if lines is not None else [int(-1)]
+ lines = [int(x) for x in lines] if lines is not None else [int(-1)]
sc = get_spark_context()
if conf is not None:
- hopDAG = sc._jvm.org.apache.sysml.api.mlcontext.MLContextUtil.getHopDAG(ml._ml, script_java, lines, conf._jconf, apply_rewrites, with_subgraph)
+ hopDAG = sc._jvm.org.apache.sysml.api.mlcontext.MLContextUtil.getHopDAG(
+ ml._ml, script_java, lines, conf._jconf, apply_rewrites, with_subgraph)
else:
- hopDAG = sc._jvm.org.apache.sysml.api.mlcontext.MLContextUtil.getHopDAG(ml._ml, script_java, lines, apply_rewrites, with_subgraph)
+ hopDAG = sc._jvm.org.apache.sysml.api.mlcontext.MLContextUtil.getHopDAG(
+ ml._ml, script_java, lines, apply_rewrites, with_subgraph)
return hopDAG
+
def dml(scriptString):
"""
Create a dml script object based on a string.
@@ -99,9 +115,12 @@ def dml(scriptString):
Instance of a script object.
"""
if not isinstance(scriptString, str):
- raise ValueError("scriptString should be a string, got %s" % type(scriptString))
+ raise ValueError(
+ "scriptString should be a string, got %s" %
+ type(scriptString))
return Script(scriptString, scriptType="dml")
+
def dmlFromResource(resourcePath):
"""
Create a dml script object based on a resource path.
@@ -117,7 +136,9 @@ def dmlFromResource(resourcePath):
Instance of a script object.
"""
if not isinstance(resourcePath, str):
- raise ValueError("resourcePath should be a string, got %s" % type(resourcePath))
+ raise ValueError(
+ "resourcePath should be a string, got %s" %
+ type(resourcePath))
return Script(resourcePath, scriptType="dml", isResource=True)
@@ -136,9 +157,12 @@ def pydml(scriptString):
Instance of a script object.
"""
if not isinstance(scriptString, str):
- raise ValueError("scriptString should be a string, got %s" % type(scriptString))
+ raise ValueError(
+ "scriptString should be a string, got %s" %
+ type(scriptString))
return Script(scriptString, scriptType="pydml")
+
def pydmlFromResource(resourcePath):
"""
Create a pydml script object based on a resource path.
@@ -154,9 +178,12 @@ def pydmlFromResource(resourcePath):
Instance of a script object.
"""
if not isinstance(resourcePath, str):
- raise ValueError("resourcePath should be a string, got %s" % type(resourcePath))
+ raise ValueError(
+ "resourcePath should be a string, got %s" %
+ type(resourcePath))
return Script(resourcePath, scriptType="pydml", isResource=True)
+
def dmlFromFile(filePath):
"""
Create a dml script object based on a file path.
@@ -172,9 +199,13 @@ def dmlFromFile(filePath):
Instance of a script object.
"""
if not isinstance(filePath, str):
- raise ValueError("filePath should be a string, got %s" % type(filePath))
- return Script(filePath, scriptType="dml", isResource=False, scriptFormat="file")
-
+ raise ValueError(
+ "filePath should be a string, got %s" %
+ type(filePath))
+ return Script(filePath, scriptType="dml",
+ isResource=False, scriptFormat="file")
+
+
def pydmlFromFile(filePath):
"""
Create a pydml script object based on a file path.
@@ -190,9 +221,12 @@ def pydmlFromFile(filePath):
Instance of a script object.
"""
if not isinstance(filePath, str):
- raise ValueError("filePath should be a string, got %s" % type(filePath))
- return Script(filePath, scriptType="pydml", isResource=False, scriptFormat="file")
-
+ raise ValueError(
+ "filePath should be a string, got %s" %
+ type(filePath))
+ return Script(filePath, scriptType="pydml",
+ isResource=False, scriptFormat="file")
+
def dmlFromUrl(url):
"""
@@ -212,6 +246,7 @@ def dmlFromUrl(url):
raise ValueError("url should be a string, got %s" % type(url))
return Script(url, scriptType="dml", isResource=False, scriptFormat="url")
+
def pydmlFromUrl(url):
"""
Create a pydml script object based on a url.
@@ -228,7 +263,9 @@ def pydmlFromUrl(url):
"""
if not isinstance(url, str):
raise ValueError("url should be a string, got %s" % type(url))
- return Script(url, scriptType="pydml", isResource=False, scriptFormat="url")
+ return Script(url, scriptType="pydml",
+ isResource=False, scriptFormat="url")
+
def _java2py(sc, obj):
""" Convert Java object to Python. """
@@ -265,6 +302,7 @@ class Matrix(object):
sc: SparkContext
SparkContext
"""
+
def __init__(self, javaMatrix, sc):
self._java_matrix = javaMatrix
self._sc = sc
@@ -297,7 +335,8 @@ class Matrix(object):
NumPy Array
A NumPy Array representing the Matrix object.
"""
- np_array = convertToNumPyArr(self._sc, self._java_matrix.toMatrixBlock())
+ np_array = convertToNumPyArr(
+ self._sc, self._java_matrix.toMatrixBlock())
return np_array
@@ -313,6 +352,7 @@ class MLResults(object):
sc: SparkContext
SparkContext
"""
+
def __init__(self, results, sc):
self._java_results = results
self._sc = sc
@@ -327,7 +367,8 @@ class MLResults(object):
outputs: string, list of strings
Output variables as defined inside the DML script.
"""
- outs = [_java2py(self._sc, self._java_results.get(out)) for out in outputs]
+ outs = [_java2py(self._sc, self._java_results.get(out))
+ for out in outputs]
if len(outs) == 1:
return outs[0]
return outs
@@ -347,70 +388,87 @@ class Script(object):
isResource: boolean
If true, scriptString is a path to a resource on the classpath
-
+
scriptFormat: string
Optional script format, either "auto" or "url" or "file" or "resource" or "string"
"""
- def __init__(self, scriptString, scriptType="dml", isResource=False, scriptFormat="auto"):
+
+ def __init__(self, scriptString, scriptType="dml",
+ isResource=False, scriptFormat="auto"):
self.sc = get_spark_context()
self.scriptString = scriptString
self.scriptType = scriptType
self.isResource = isResource
if scriptFormat != "auto":
if scriptFormat == "url" and self.scriptType == "dml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromUrl(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromUrl(
+ scriptString)
elif scriptFormat == "url" and self.scriptType == "pydml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromUrl(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromUrl(
+ scriptString)
elif scriptFormat == "file" and self.scriptType == "dml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromFile(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromFile(
+ scriptString)
elif scriptFormat == "file" and self.scriptType == "pydml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromFile(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromFile(
+ scriptString)
elif isResource and self.scriptType == "dml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromResource(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromResource(
+ scriptString)
elif isResource and self.scriptType == "pydml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromResource(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromResource(
+ scriptString)
elif scriptFormat == "string" and self.scriptType == "dml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dml(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dml(
+ scriptString)
elif scriptFormat == "string" and self.scriptType == "pydml":
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydml(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydml(
+ scriptString)
else:
raise ValueError('Unsupported script format' + scriptFormat)
elif self.scriptType == "dml":
if scriptString.endswith(".dml"):
if scriptString.startswith("http"):
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromUrl(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromUrl(
+ scriptString)
elif os.path.exists(scriptString):
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromFile(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromFile(
+ scriptString)
elif self.isResource == True:
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromResource(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dmlFromResource(
+ scriptString)
else:
raise ValueError("path: %s does not exist" % scriptString)
else:
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dml(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.dml(
+ scriptString)
elif self.scriptType == "pydml":
if scriptString.endswith(".pydml"):
if scriptString.startswith("http"):
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromUrl(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromUrl(
+ scriptString)
elif os.path.exists(scriptString):
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromFile(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromFile(
+ scriptString)
elif self.isResource == True:
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromResource(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydmlFromResource(
+ scriptString)
else:
raise ValueError("path: %s does not exist" % scriptString)
else:
- self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydml(scriptString)
+ self.script_java = self.sc._jvm.org.apache.sysml.api.mlcontext.ScriptFactory.pydml(
+ scriptString)
-
def getScriptString(self):
"""
Obtain the script string (in unicode).
"""
return self.script_java.getScriptString()
-
+
def setScriptString(self, scriptString):
"""
Set the script string.
-
+
Parameters
----------
scriptString: string
@@ -438,80 +496,80 @@ class Script(object):
"""
self.script_java.clearIOS()
return self
-
+
def clearIO(self):
"""
Clear the inputs and outputs, but not the symbol table.
"""
self.script_java.clearIO()
return self
-
+
def clearAll(self):
"""
Clear the script string, inputs, outputs, and symbol table.
"""
self.script_java.clearAll()
return self
-
+
def clearInputs(self):
"""
Clear the inputs.
"""
self.script_java.clearInputs()
return self
-
+
def clearOutputs(self):
"""
Clear the outputs.
"""
self.script_java.clearOutputs()
return self
-
+
def clearSymbolTable(self):
"""
Clear the symbol table.
"""
self.script_java.clearSymbolTable()
return self
-
+
def results(self):
"""
Obtain the results of the script execution.
"""
return MLResults(self.script_java.results(), self.sc)
-
+
def getResults(self):
"""
Obtain the results of the script execution.
"""
return MLResults(self.script_java.getResults(), self.sc)
-
+
def setResults(self, results):
"""
Set the results of the script execution.
"""
self.script_java.setResults(results._java_results)
return self
-
+
def isDML(self):
"""
Is the script type DML?
"""
return self.script_java.isDML()
-
+
def isPYDML(self):
"""
Is the script type DML?
"""
return self.script_java.isPYDML()
-
+
def getScriptExecutionString(self):
"""
Generate the script execution string, which adds read/load/write/save
statements to the beginning and end of the script to execute.
"""
- return self.script_java.getScriptExecutionString()
-
+ return self.script_java.getScriptExecutionString()
+
def __repr__(self):
return "Script"
@@ -528,56 +586,56 @@ class Script(object):
Display the script inputs.
"""
return self.script_java.displayInputs()
-
+
def displayOutputs(self):
"""
Display the script outputs.
"""
return self.script_java.displayOutputs()
-
+
def displayInputParameters(self):
"""
Display the script input parameters.
"""
return self.script_java.displayInputParameters()
-
+
def displayInputVariables(self):
"""
Display the script input variables.
"""
return self.script_java.displayInputVariables()
-
+
def displayOutputVariables(self):
"""
Display the script output variables.
"""
return self.script_java.displayOutputVariables()
-
+
def displaySymbolTable(self):
"""
Display the script symbol table.
"""
return self.script_java.displaySymbolTable()
-
+
def getName(self):
"""
Obtain the script name.
"""
return self.script_java.getName()
-
+
def setName(self, name):
"""
Set the script name.
"""
self.script_java.setName(name)
return self
-
+
def getScriptType(self):
"""
Obtain the script type.
"""
return self.scriptType
-
+
def input(self, *args, **kwargs):
"""
Parameters
@@ -606,9 +664,11 @@ class Script(object):
if isinstance(val, py4j.java_gateway.JavaObject):
py4j.java_gateway.get_method(self.script_java, "in")(key, val)
else:
- py4j.java_gateway.get_method(self.script_java, "in")(key, _py2java(self.sc, val))
-
-
+ py4j.java_gateway.get_method(
+ self.script_java, "in")(
+ key, _py2java(
+ self.sc, val))
+
def output(self, *names):
"""
Parameters
@@ -630,17 +690,20 @@ class MLContext(object):
sc: SparkContext or SparkSession
An instance of pyspark.SparkContext or pyspark.sql.SparkSession.
"""
+
def __init__(self, sc):
if isinstance(sc, pyspark.sql.session.SparkSession):
sc = sc._sc
elif not isinstance(sc, SparkContext):
- raise ValueError("Expected sc to be a SparkContext or SparkSession, got " % str(type(sc)))
+ raise ValueError(
+ "Expected sc to be a SparkContext or SparkSession, got " % str(
+ type(sc)))
self._sc = sc
self._ml = createJavaObject(sc, 'mlcontext')
def __repr__(self):
return "MLContext"
-
+
def execute(self, script):
"""
Execute a DML / PyDML script.
@@ -688,7 +751,7 @@ class MLContext(object):
"""
self._ml.setGPU(bool(enable))
return self
-
+
def setForceGPU(self, enable):
"""
Whether or not to force the usage of GPU operators.
@@ -699,7 +762,7 @@ class MLContext(object):
"""
self._ml.setForceGPU(bool(enable))
return self
-
+
def setStatisticsMaxHeavyHitters(self, maxHeavyHitters):
"""
The maximum number of heavy hitters that are printed as part of the statistics.
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/mllearn/__init__.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/mllearn/__init__.py b/src/main/python/systemml/mllearn/__init__.py
index 907ffb2..eda2197 100644
--- a/src/main/python/systemml/mllearn/__init__.py
+++ b/src/main/python/systemml/mllearn/__init__.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,7 +17,7 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
"""
===================
http://git-wip-us.apache.org/repos/asf/systemml/blob/9e7ee19a/src/main/python/systemml/mllearn/estimators.py
----------------------------------------------------------------------
diff --git a/src/main/python/systemml/mllearn/estimators.py b/src/main/python/systemml/mllearn/estimators.py
index 52fe9cc..0b3de41 100644
--- a/src/main/python/systemml/mllearn/estimators.py
+++ b/src/main/python/systemml/mllearn/estimators.py
@@ -1,4 +1,4 @@
-#-------------------------------------------------------------
+# -------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,9 +17,15 @@
# specific language governing permissions and limitations
# under the License.
#
-#-------------------------------------------------------------
+# -------------------------------------------------------------
-__all__ = ['LinearRegression', 'LogisticRegression', 'SVM', 'NaiveBayes', 'Caffe2DML', 'Keras2DML']
+__all__ = [
+ 'LinearRegression',
+ 'LogisticRegression',
+ 'SVM',
+ 'NaiveBayes',
+ 'Caffe2DML',
+ 'Keras2DML']
import numpy as np
from pyspark.ml import Estimator
@@ -37,15 +43,17 @@ import math
from ..converters import *
from ..classloader import *
+
def assemble(sparkSession, pdf, inputCols, outputCol):
tmpDF = sparkSession.createDataFrame(pdf, list(pdf.columns))
assembler = VectorAssembler(inputCols=list(inputCols), outputCol=outputCol)
return assembler.transform(tmpDF)
+
class BaseSystemMLEstimator(Estimator):
features_col = 'features'
label_col = 'label'
-
+
def set_features_col(self, colName):
"""
Sets the default column name for features of PySpark DataFrame.
@@ -76,7 +84,7 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setGPU(enable)
return self
-
+
def setForceGPU(self, enable):
"""
Whether or not to force the usage of GPU operators.
@@ -87,7 +95,7 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setForceGPU(enable)
return self
-
+
def setExplain(self, explain):
"""
Explanation about the program. Mainly intended for developers.
@@ -98,11 +106,11 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setExplain(explain)
return self
-
+
def setExplainLevel(self, explainLevel):
"""
Set explain level. Mainly intended for developers.
-
+
Parameters
----------
explainLevel: string
@@ -111,7 +119,7 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setExplainLevel(explainLevel)
return self
-
+
def setStatistics(self, statistics):
"""
Whether or not to output statistics (such as execution time, elapsed time)
@@ -123,7 +131,7 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setStatistics(statistics)
return self
-
+
def setStatisticsMaxHeavyHitters(self, maxHeavyHitters):
"""
The maximum number of heavy hitters that are printed as part of the statistics.
@@ -134,7 +142,7 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setStatisticsMaxHeavyHitters(maxHeavyHitters)
return self
-
+
def setConfigProperty(self, propertyName, propertyValue):
"""
Set configuration property, such as setConfigProperty("sysml.localtmpdir", "/tmp/systemml").
@@ -146,7 +154,7 @@ class BaseSystemMLEstimator(Estimator):
"""
self.estimator.setConfigProperty(propertyName, propertyValue)
return self
-
+
def _fit_df(self):
global default_jvm_stdout, default_jvm_stdout_parallel_flush
try:
@@ -157,28 +165,30 @@ class BaseSystemMLEstimator(Estimator):
self.model = self.estimator.fit(self.X._jdf)
except Py4JError:
traceback.print_exc()
-
+
def fit_df(self, X):
self.X = X
self._fit_df()
self.X = None
return self
-
+
def _fit_numpy(self):
global default_jvm_stdout, default_jvm_stdout_parallel_flush
try:
- if type(self.y) == np.ndarray and len(self.y.shape) == 1:
+ if isinstance(self.y, np.ndarray) and len(self.y.shape) == 1:
# Since we know that mllearn always needs a column vector
self.y = np.matrix(self.y).T
y_mb = convertToMatrixBlock(self.sc, self.y)
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
- self.model = self.estimator.fit(convertToMatrixBlock(self.sc, self.X), y_mb)
+ self.model = self.estimator.fit(
+ convertToMatrixBlock(self.sc, self.X), y_mb)
else:
- self.model = self.estimator.fit(convertToMatrixBlock(self.sc, self.X), y_mb)
+ self.model = self.estimator.fit(
+ convertToMatrixBlock(self.sc, self.X), y_mb)
except Py4JError:
traceback.print_exc()
-
+
def fit_numpy(self, X, y):
self.X = X
self.y = y
@@ -198,7 +208,7 @@ class BaseSystemMLEstimator(Estimator):
except Py4JError:
traceback.print_exc()
return self
-
+
# Returns a model after calling fit(df) on Estimator object on JVM
def _fit(self, X):
"""
@@ -208,10 +218,12 @@ class BaseSystemMLEstimator(Estimator):
----------
X: PySpark DataFrame that contain the columns features_col (default: 'features') and label_col (default: 'label')
"""
- if hasattr(X, '_jdf') and self.features_col in X.columns and self.label_col in X.columns:
+ if hasattr(
+ X, '_jdf') and self.features_col in X.columns and self.label_col in X.columns:
return self.fit_df(X)
else:
- raise Exception('Incorrect usage: Expected dataframe as input with features/label as columns')
+ raise Exception(
+ 'Incorrect usage: Expected dataframe as input with features/label as columns')
def fit(self, X, y=None, params=None):
"""
@@ -228,7 +240,8 @@ class BaseSystemMLEstimator(Estimator):
return self.fit_file(X, y)
elif isinstance(X, SUPPORTED_TYPES) and isinstance(y, SUPPORTED_TYPES):
# Donot encode if y is a numpy matrix => useful for segmentation
- skipEncodingY = len(y.shape) == 2 and y.shape[0] != 1 and y.shape[1] != 1
+ skipEncodingY = len(
+ y.shape) == 2 and y.shape[0] != 1 and y.shape[1] != 1
y = y if skipEncodingY else self.encode(y)
if self.transferUsingDF:
pdfX = convertToPandasDF(X)
@@ -239,7 +252,13 @@ class BaseSystemMLEstimator(Estimator):
raise Exception('Number of rows of X and y should match')
colNames = pdfX.columns
pdfX[self.label_col] = pdfY[pdfY.columns[0]]
- df = assemble(self.sparkSession, pdfX, colNames, self.features_col).select(self.features_col, self.label_col)
+ df = assemble(
+ self.sparkSession,
+ pdfX,
+ colNames,
+ self.features_col).select(
+ self.features_col,
+ self.label_col)
self.fit_df(df)
else:
numColsy = getNumCols(y)
@@ -254,7 +273,7 @@ class BaseSystemMLEstimator(Estimator):
def transform(self, X):
return self.predict(X)
-
+
def _convertPythonXToJavaObject(self, X):
"""
Converts the input python object X to a java-side object (either MatrixBlock or Java DataFrame)
@@ -265,7 +284,12 @@ class BaseSystemMLEstimator(Estimator):
"""
if isinstance(X, SUPPORTED_TYPES) and self.transferUsingDF:
pdfX = convertToPandasDF(X)
- df = assemble(self.sparkSession, pdfX, pdfX.columns, self.features_col).select(self.features_col)
+ df = assemble(
+ self.sparkSession,
+ pdfX,
+ pdfX.columns,
+ self.features_col).select(
+ self.features_col)
return df._jdf
elif isinstance(X, SUPPORTED_TYPES):
return convertToMatrixBlock(self.sc, X)
@@ -273,12 +297,13 @@ class BaseSystemMLEstimator(Estimator):
# No need to assemble as input DF is likely coming via MLPipeline
return X._jdf
elif hasattr(X, '_jdf'):
- assembler = VectorAssembler(inputCols=X.columns, outputCol=self.features_col)
+ assembler = VectorAssembler(
+ inputCols=X.columns, outputCol=self.features_col)
df = assembler.transform(X)
return df._jdf
else:
raise Exception('Unsupported input type')
-
+
def _convertJavaOutputToPythonObject(self, X, output):
"""
Converts the a java-side object output (either MatrixBlock or Java DataFrame) to a python object (based on the type of X).
@@ -300,7 +325,7 @@ class BaseSystemMLEstimator(Estimator):
return retDF.sort('__INDEX')
else:
raise Exception('Unsupported input type')
-
+
def predict_proba(self, X):
"""
Invokes the transform_probability method on Estimator object on JVM if X and y are on of the supported data types
@@ -314,7 +339,8 @@ class BaseSystemMLEstimator(Estimator):
if hasattr(X, '_jdf'):
return self.predict(X)
elif self.transferUsingDF:
- raise ValueError('The parameter transferUsingDF is not valid for the method predict_proba')
+ raise ValueError(
+ 'The parameter transferUsingDF is not valid for the method predict_proba')
try:
if self.estimator is not None and self.model is not None:
self.estimator.copyProperties(self.model)
@@ -326,13 +352,16 @@ class BaseSystemMLEstimator(Estimator):
jX = self._convertPythonXToJavaObject(X)
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
- return self._convertJavaOutputToPythonObject(X, self.model.transform_probability(jX))
+ return self._convertJavaOutputToPythonObject(
+ X, self.model.transform_probability(jX))
else:
- return self._convertJavaOutputToPythonObject(X, self.model.transform_probability(jX))
+ return self._convertJavaOutputToPythonObject(
+ X, self.model.transform_probability(jX))
except Py4JError:
traceback.print_exc()
-
- # Returns either a DataFrame or MatrixBlock after calling transform(X:MatrixBlock, y:MatrixBlock) on Model object on JVM
+
+ # Returns either a DataFrame or MatrixBlock after calling
+ # transform(X:MatrixBlock, y:MatrixBlock) on Model object on JVM
def predict(self, X):
"""
Invokes the transform method on Estimator object on JVM if X and y are on of the supported data types
@@ -353,20 +382,23 @@ class BaseSystemMLEstimator(Estimator):
jX = self._convertPythonXToJavaObject(X)
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
- ret = self._convertJavaOutputToPythonObject(X, self.model.transform(jX))
+ ret = self._convertJavaOutputToPythonObject(
+ X, self.model.transform(jX))
else:
- ret = self._convertJavaOutputToPythonObject(X, self.model.transform(jX))
+ ret = self._convertJavaOutputToPythonObject(
+ X, self.model.transform(jX))
return self.decode(ret) if isinstance(X, SUPPORTED_TYPES) else ret
except Py4JError:
traceback.print_exc()
+
class BaseSystemMLClassifier(BaseSystemMLEstimator):
def encode(self, y):
self.le = LabelEncoder()
self.le.fit(y)
return self.le.transform(y) + 1
-
+
def decode(self, y):
if not hasattr(self, 'le'):
self.le = None
@@ -375,14 +407,14 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
if self.le is not None:
return self.le.inverse_transform(np.asarray(y - 1, dtype=int))
elif self.labelMap is not None:
- return [ self.labelMap[int(i)] for i in y ]
+ return [self.labelMap[int(i)] for i in y]
else:
return y
-
+
def predict(self, X):
predictions = super(BaseSystemMLClassifier, self).predict(X)
from pyspark.sql.dataframe import DataFrame as df
- if type(predictions) == df:
+ if isinstance(predictions, df):
return predictions
else:
try:
@@ -390,7 +422,7 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
except ValueError:
print(type(predictions))
return np.asarray(predictions, dtype='str')
-
+
def score(self, X, y):
"""
Scores the predicted value with ground truth 'y'
@@ -404,8 +436,9 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
if np.issubdtype(predictions.dtype.type, np.number):
return accuracy_score(y, predictions)
else:
- return accuracy_score(np.asarray(y, dtype='str'), np.asarray(predictions, dtype='str'))
-
+ return accuracy_score(np.asarray(
+ y, dtype='str'), np.asarray(predictions, dtype='str'))
+
def loadLabels(self, file_path):
createJavaObject(self.sc, 'dummy')
utilObj = self.sc._jvm.org.apache.sysml.api.ml.Utils()
@@ -417,10 +450,10 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
for i in range(len(keys)):
self.labelMap[int(keys[i])] = values[i]
# self.encode(classes) # Giving incorrect results
-
+
def load(self, weights, sep='/', eager=False):
"""
- Load a pretrained model.
+ Load a pretrained model.
Parameters
----------
@@ -436,11 +469,11 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
else:
self.model.load(self.sc._jsc, weights, sep, eager)
self.loadLabels(weights + '/labels.txt')
-
+
def save(self, outputDir, format='binary', sep='/'):
"""
Save a trained model.
-
+
Parameters
----------
outputDir: Directory to save the model to
@@ -448,7 +481,7 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
sep: seperator to use (default: '/')
"""
global default_jvm_stdout, default_jvm_stdout_parallel_flush
- if self.model != None:
+ if self.model is not None:
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
self.model.save(self.sc._jsc, outputDir, format, sep)
@@ -462,21 +495,26 @@ class BaseSystemMLClassifier(BaseSystemMLEstimator):
labelMapping = self.labelMap
if labelMapping is not None:
- lStr = [ [ int(k), str(labelMapping[k]) ] for k in labelMapping ]
+ lStr = [[int(k), str(labelMapping[k])] for k in labelMapping]
df = self.sparkSession.createDataFrame(lStr)
- df.write.csv(outputDir + sep + 'labels.txt', mode='overwrite', header=False)
+ df.write.csv(
+ outputDir + sep + 'labels.txt',
+ mode='overwrite',
+ header=False)
else:
- raise Exception('Cannot save as you need to train the model first using fit')
+ raise Exception(
+ 'Cannot save as you need to train the model first using fit')
return self
+
class BaseSystemMLRegressor(BaseSystemMLEstimator):
def encode(self, y):
return y
-
+
def decode(self, y):
return y
-
+
def score(self, X, y):
"""
Scores the predicted value with ground truth 'y'
@@ -487,10 +525,10 @@ class BaseSystemMLRegressor(BaseSystemMLEstimator):
y: NumPy ndarray, Pandas DataFrame, scipy sparse matrix
"""
return r2_score(y, self.predict(X), multioutput='variance_weighted')
-
+
def load(self, weights=None, sep='/', eager=False):
"""
- Load a pretrained model.
+ Load a pretrained model.
Parameters
----------
@@ -509,7 +547,7 @@ class BaseSystemMLRegressor(BaseSystemMLEstimator):
def save(self, outputDir, format='binary', sep='/'):
"""
Save a trained model.
-
+
Parameters
----------
outputDir: Directory to save the model to
@@ -517,14 +555,15 @@ class BaseSystemMLRegressor(BaseSystemMLEstimator):
sep: seperator to use (default: '/')
"""
global default_jvm_stdout, default_jvm_stdout_parallel_flush
- if self.model != None:
+ if self.model is not None:
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
self.model.save(outputDir, format, sep)
else:
self.model.save(outputDir, format, sep)
else:
- raise Exception('Cannot save as you need to train the model first using fit')
+ raise Exception(
+ 'Cannot save as you need to train the model first using fit')
return self
@@ -534,9 +573,9 @@ class LogisticRegression(BaseSystemMLClassifier):
Examples
--------
-
+
Scikit-learn way
-
+
>>> from sklearn import datasets, neighbors
>>> from systemml.mllearn import LogisticRegression
>>> from pyspark.sql import SparkSession
@@ -551,9 +590,9 @@ class LogisticRegression(BaseSystemMLClassifier):
>>> y_test = y_digits[.9 * n_samples:]
>>> logistic = LogisticRegression(sparkSession)
>>> print('LogisticRegression score: %f' % logistic.fit(X_train, y_train).score(X_test, y_test))
-
+
MLPipeline way
-
+
>>> from pyspark.ml import Pipeline
>>> from systemml.mllearn import LogisticRegression
>>> from pyspark.ml.feature import HashingTF, Tokenizer
@@ -585,13 +624,14 @@ class LogisticRegression(BaseSystemMLClassifier):
>>> (15L, "apache hadoop")], ["id", "text"])
>>> prediction = model.transform(test)
>>> prediction.show()
-
+
"""
-
- def __init__(self, sparkSession, penalty='l2', fit_intercept=True, normalize=False, max_iter=100, max_inner_iter=0, tol=0.000001, C=1.0, solver='newton-cg', transferUsingDF=False):
+
+ def __init__(self, sparkSession, penalty='l2', fit_intercept=True, normalize=False, max_iter=100,
+ max_inner_iter=0, tol=0.000001, C=1.0, solver='newton-cg', transferUsingDF=False):
"""
Performs both binomial and multinomial logistic regression.
-
+
Parameters
----------
sparkSession: PySpark SparkSession
@@ -608,30 +648,33 @@ class LogisticRegression(BaseSystemMLClassifier):
self.sc = sparkSession._sc
createJavaObject(self.sc, 'dummy')
self.uid = "logReg"
- self.estimator = self.sc._jvm.org.apache.sysml.api.ml.LogisticRegression(self.uid, self.sc._jsc.sc())
+ self.estimator = self.sc._jvm.org.apache.sysml.api.ml.LogisticRegression(
+ self.uid, self.sc._jsc.sc())
self.estimator.setMaxOuterIter(max_iter)
self.estimator.setMaxInnerIter(max_inner_iter)
reg = 0.0 if C == float("inf") else 1.0 / C
- icpt = 2 if fit_intercept == True and normalize == True else int(fit_intercept)
+ icpt = 2 if fit_intercept == True and normalize == True else int(
+ fit_intercept)
self.estimator.setRegParam(reg)
self.estimator.setTol(tol)
self.estimator.setIcpt(icpt)
self.transferUsingDF = transferUsingDF
self.setOutputRawPredictionsToFalse = True
- self.model = self.sc._jvm.org.apache.sysml.api.ml.LogisticRegressionModel(self.estimator)
+ self.model = self.sc._jvm.org.apache.sysml.api.ml.LogisticRegressionModel(
+ self.estimator)
if penalty != 'l2':
raise Exception('Only l2 penalty is supported')
if solver != 'newton-cg':
raise Exception('Only newton-cg solver supported')
-
+
class LinearRegression(BaseSystemMLRegressor):
"""
Performs linear regression to model the relationship between one numerical response variable and one or more explanatory (feature) variables.
-
+
Examples
--------
-
+
>>> import numpy as np
>>> from sklearn import datasets
>>> from systemml.mllearn import LinearRegression
@@ -652,11 +695,11 @@ class LinearRegression(BaseSystemMLRegressor):
>>> regr.fit(diabetes_X_train, diabetes_y_train)
>>> # The mean square error
>>> print("Residual sum of squares: %.2f" % np.mean((regr.predict(diabetes_X_test) - diabetes_y_test) ** 2))
-
+
"""
-
-
- def __init__(self, sparkSession, fit_intercept=True, normalize=False, max_iter=100, tol=0.000001, C=float("inf"), solver='newton-cg', transferUsingDF=False):
+
+ def __init__(self, sparkSession, fit_intercept=True, normalize=False, max_iter=100,
+ tol=0.000001, C=float("inf"), solver='newton-cg', transferUsingDF=False):
"""
Performs linear regression to model the relationship between one numerical response variable and one or more explanatory (feature) variables.
@@ -678,18 +721,21 @@ class LinearRegression(BaseSystemMLRegressor):
createJavaObject(self.sc, 'dummy')
self.uid = "lr"
if solver == 'newton-cg' or solver == 'direct-solve':
- self.estimator = self.sc._jvm.org.apache.sysml.api.ml.LinearRegression(self.uid, self.sc._jsc.sc(), solver)
+ self.estimator = self.sc._jvm.org.apache.sysml.api.ml.LinearRegression(
+ self.uid, self.sc._jsc.sc(), solver)
else:
raise Exception('Only newton-cg solver supported')
self.estimator.setMaxIter(max_iter)
reg = 0.0 if C == float("inf") else 1.0 / C
- icpt = 2 if fit_intercept == True and normalize == True else int(fit_intercept)
+ icpt = 2 if fit_intercept == True and normalize == True else int(
+ fit_intercept)
self.estimator.setRegParam(reg)
self.estimator.setTol(tol)
self.estimator.setIcpt(icpt)
self.transferUsingDF = transferUsingDF
self.setOutputRawPredictionsToFalse = False
- self.model = self.sc._jvm.org.apache.sysml.api.ml.LinearRegressionModel(self.estimator)
+ self.model = self.sc._jvm.org.apache.sysml.api.ml.LinearRegressionModel(
+ self.estimator)
class SVM(BaseSystemMLClassifier):
@@ -698,14 +744,14 @@ class SVM(BaseSystemMLClassifier):
Examples
--------
-
+
>>> from sklearn import datasets, neighbors
>>> from systemml.mllearn import SVM
>>> from pyspark.sql import SparkSession
>>> sparkSession = SparkSession.builder.getOrCreate()
>>> digits = datasets.load_digits()
>>> X_digits = digits.data
- >>> y_digits = digits.target
+ >>> y_digits = digits.target
>>> n_samples = len(X_digits)
>>> X_train = X_digits[:.9 * n_samples]
>>> y_train = y_digits[:.9 * n_samples]
@@ -713,11 +759,11 @@ class SVM(BaseSystemMLClassifier):
>>> y_test = y_digits[.9 * n_samples:]
>>> svm = SVM(sparkSession, is_multi_class=True)
>>> print('LogisticRegression score: %f' % svm.fit(X_train, y_train).score(X_test, y_test))
-
- """
+ """
- def __init__(self, sparkSession, fit_intercept=True, normalize=False, max_iter=100, tol=0.000001, C=1.0, is_multi_class=False, transferUsingDF=False):
+ def __init__(self, sparkSession, fit_intercept=True, normalize=False, max_iter=100,
+ tol=0.000001, C=1.0, is_multi_class=False, transferUsingDF=False):
"""
Performs both binary-class and multiclass SVM (Support Vector Machines).
@@ -736,18 +782,22 @@ class SVM(BaseSystemMLClassifier):
self.uid = "svm"
createJavaObject(self.sc, 'dummy')
self.is_multi_class = is_multi_class
- self.estimator = self.sc._jvm.org.apache.sysml.api.ml.SVM(self.uid, self.sc._jsc.sc(), is_multi_class)
+ self.estimator = self.sc._jvm.org.apache.sysml.api.ml.SVM(
+ self.uid, self.sc._jsc.sc(), is_multi_class)
self.estimator.setMaxIter(max_iter)
if C <= 0:
raise Exception('C has to be positive')
reg = 0.0 if C == float("inf") else 1.0 / C
- icpt = 2 if fit_intercept == True and normalize == True else int(fit_intercept)
+ icpt = 2 if fit_intercept == True and normalize == True else int(
+ fit_intercept)
self.estimator.setRegParam(reg)
self.estimator.setTol(tol)
self.estimator.setIcpt(icpt)
self.transferUsingDF = transferUsingDF
self.setOutputRawPredictionsToFalse = False
- self.model = self.sc._jvm.org.apache.sysml.api.ml.SVMModel(self.estimator, self.is_multi_class)
+ self.model = self.sc._jvm.org.apache.sysml.api.ml.SVMModel(
+ self.estimator, self.is_multi_class)
+
class NaiveBayes(BaseSystemMLClassifier):
"""
@@ -755,7 +805,7 @@ class NaiveBayes(BaseSystemMLClassifier):
Examples
--------
-
+
>>> from sklearn.datasets import fetch_20newsgroups
>>> from sklearn.feature_extraction.text import TfidfVectorizer
>>> from systemml.mllearn import NaiveBayes
@@ -775,7 +825,7 @@ class NaiveBayes(BaseSystemMLClassifier):
>>> metrics.f1_score(newsgroups_test.target, pred, average='weighted')
"""
-
+
def __init__(self, sparkSession, laplace=1.0, transferUsingDF=False):
"""
Performs Naive Bayes.
@@ -789,19 +839,22 @@ class NaiveBayes(BaseSystemMLClassifier):
self.sc = sparkSession._sc
self.uid = "nb"
createJavaObject(self.sc, 'dummy')
- self.estimator = self.sc._jvm.org.apache.sysml.api.ml.NaiveBayes(self.uid, self.sc._jsc.sc())
+ self.estimator = self.sc._jvm.org.apache.sysml.api.ml.NaiveBayes(
+ self.uid, self.sc._jsc.sc())
self.estimator.setLaplace(laplace)
self.transferUsingDF = transferUsingDF
self.setOutputRawPredictionsToFalse = False
- self.model = self.sc._jvm.org.apache.sysml.api.ml.NaiveBayesModel(self.estimator)
+ self.model = self.sc._jvm.org.apache.sysml.api.ml.NaiveBayesModel(
+ self.estimator)
+
class Caffe2DML(BaseSystemMLClassifier):
"""
Performs training/prediction for a given caffe network.
-
+
Examples
--------
-
+
>>> from systemml.mllearn import Caffe2DML
>>> from mlxtend.data import mnist_data
>>> import numpy as np
@@ -815,9 +868,11 @@ class Caffe2DML(BaseSystemMLClassifier):
>>> caffe2DML = Caffe2DML(spark, 'lenet_solver.proto').set(max_iter=500)
>>> caffe2DML.fit(X, y)
"""
- def __init__(self, sparkSession, solver, input_shape, transferUsingDF=False):
+
+ def __init__(self, sparkSession, solver, input_shape,
+ transferUsingDF=False):
"""
- Performs training/prediction for a given caffe network.
+ Performs training/prediction for a given caffe network.
Parameters
----------
@@ -833,14 +888,19 @@ class Caffe2DML(BaseSystemMLClassifier):
self.model = None
if len(input_shape) != 3:
raise ValueError('Expected input_shape as list of 3 element')
- solver = self.sc._jvm.org.apache.sysml.api.dl.Utils.readCaffeSolver(solver)
- self.estimator = self.sc._jvm.org.apache.sysml.api.dl.Caffe2DML(self.sc._jsc.sc(), solver, str(input_shape[0]), str(input_shape[1]), str(input_shape[2]))
+ solver = self.sc._jvm.org.apache.sysml.api.dl.Utils.readCaffeSolver(
+ solver)
+ self.estimator = self.sc._jvm.org.apache.sysml.api.dl.Caffe2DML(
+ self.sc._jsc.sc(), solver, str(
+ input_shape[0]), str(
+ input_shape[1]), str(
+ input_shape[2]))
self.transferUsingDF = transferUsingDF
self.setOutputRawPredictionsToFalse = False
def load(self, weights=None, sep='/', ignore_weights=None, eager=False):
"""
- Load a pretrained model.
+ Load a pretrained model.
Parameters
----------
@@ -852,7 +912,8 @@ class Caffe2DML(BaseSystemMLClassifier):
global default_jvm_stdout, default_jvm_stdout_parallel_flush
self.weights = weights
self.estimator.setInput("$weights", str(weights))
- self.model = self.sc._jvm.org.apache.sysml.api.dl.Caffe2DMLModel(self.estimator)
+ self.model = self.sc._jvm.org.apache.sysml.api.dl.Caffe2DMLModel(
+ self.estimator)
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
self.model.load(self.sc._jsc, weights, sep, eager)
@@ -861,11 +922,12 @@ class Caffe2DML(BaseSystemMLClassifier):
self.loadLabels(weights + '/labels.txt')
if ignore_weights is not None:
self.estimator.setWeightsToIgnore(ignore_weights)
-
- def set(self, debug=None, train_algo=None, test_algo=None, parallel_batches=None, output_activations=None, perform_one_hot_encoding=None, parfor_parameters=None):
+
+ def set(self, debug=None, train_algo=None, test_algo=None, parallel_batches=None,
+ output_activations=None, perform_one_hot_encoding=None, parfor_parameters=None):
"""
Set input to Caffe2DML
-
+
Parameters
----------
debug: to add debugging DML code such as classification report, print DML script, etc (default: False)
@@ -876,37 +938,55 @@ class Caffe2DML(BaseSystemMLClassifier):
perform_one_hot_encoding: should perform one-hot encoding in DML using table function (default: False)
parfor_parameters: dictionary for parfor parameters when using allreduce-style algorithms (default: "")
"""
- if debug is not None: self.estimator.setInput("$debug", str(debug).upper())
- if train_algo is not None: self.estimator.setInput("$train_algo", str(train_algo).lower())
- if test_algo is not None: self.estimator.setInput("$test_algo", str(test_algo).lower())
- if parallel_batches is not None: self.estimator.setInput("$parallel_batches", str(parallel_batches))
- if output_activations is not None: self.estimator.setInput("$output_activations", str(output_activations))
- if perform_one_hot_encoding is not None: self.estimator.setInput("$perform_one_hot_encoding", str(perform_one_hot_encoding).lower())
+ if debug is not None:
+ self.estimator.setInput("$debug", str(debug).upper())
+ if train_algo is not None:
+ self.estimator.setInput("$train_algo", str(train_algo).lower())
+ if test_algo is not None:
+ self.estimator.setInput("$test_algo", str(test_algo).lower())
+ if parallel_batches is not None:
+ self.estimator.setInput("$parallel_batches", str(parallel_batches))
+ if output_activations is not None:
+ self.estimator.setInput(
+ "$output_activations",
+ str(output_activations))
+ if perform_one_hot_encoding is not None:
+ self.estimator.setInput(
+ "$perform_one_hot_encoding",
+ str(perform_one_hot_encoding).lower())
if parfor_parameters is not None:
if isinstance(parfor_parameters, dict):
# Convert dictionary to comma-separated list
- parfor_parameters = ''.join([ ', ' + str(k) + '=' + str(v) for k, v in parfor_parameters.items()]) if len(parfor_parameters) > 0 else ''
- self.estimator.setInput("$parfor_parameters", parfor_parameters)
+ parfor_parameters = ''.join(
+ [
+ ', ' +
+ str(k) +
+ '=' +
+ str(v) for k,
+ v in parfor_parameters.items()]) if len(parfor_parameters) > 0 else ''
+ self.estimator.setInput(
+ "$parfor_parameters", parfor_parameters)
else:
- raise TypeError("parfor_parameters should be a dictionary")
+ raise TypeError("parfor_parameters should be a dictionary")
return self
-
+
def summary(self):
"""
Print the summary of the network
"""
import pyspark
global default_jvm_stdout, default_jvm_stdout_parallel_flush
- if type(self.sparkSession) == pyspark.sql.session.SparkSession:
+ if isinstance(self.sparkSession, pyspark.sql.session.SparkSession):
if default_jvm_stdout:
with jvm_stdout(parallel_flush=default_jvm_stdout_parallel_flush):
self.estimator.summary(self.sparkSession._jsparkSession)
else:
self.estimator.summary(self.sparkSession._jsparkSession)
else:
- raise TypeError('Please use spark session of type pyspark.sql.session.SparkSession in the constructor')
-
-
+ raise TypeError(
+ 'Please use spark session of type pyspark.sql.session.SparkSession in the constructor')
+
+
class Keras2DML(Caffe2DML):
"""
Peforms training/prediction for a given keras model.
@@ -914,7 +994,8 @@ class Keras2DML(Caffe2DML):
"""
- def __init__(self, sparkSession, keras_model, input_shape, transferUsingDF=False, load_keras_weights=True, weights=None, labels=None, batch_size=64, max_iter=2000, test_iter=10, test_interval=500, display=100, lr_policy="step", weight_decay=5e-4, regularization_type="L2"):
+ def __init__(self, sparkSession, keras_model, input_shape, transferUsingDF=False, load_keras_weights=True, weights=None, labels=None,
+ batch_size=64, max_iter=2000, test_iter=10, test_interval=500, display=100, lr_policy="step", weight_decay=5e-4, regularization_type="L2"):
"""
Performs training/prediction for a given keras model.
@@ -936,9 +1017,9 @@ class Keras2DML(Caffe2DML):
weight_decay: regularation strength (default: 5e-4)
regularization_type: regularization type (default: "L2")
"""
- from .keras2caffe import *
+ from .keras2caffe import convertKerasToCaffeNetwork, convertKerasToCaffeSolver
import tempfile
- if type(keras_model) == keras.models.Sequential:
+ if isinstance(keras_model, keras.models.Sequential):
# Convert the sequential model to functional model
if keras_model.model is None:
keras_model.build()
@@ -946,19 +1027,47 @@ class Keras2DML(Caffe2DML):
self.name = keras_model.name
createJavaObject(sparkSession._sc, 'dummy')
if not hasattr(keras_model, 'optimizer'):
- keras_model.compile(loss='categorical_crossentropy', optimizer=keras.optimizers.SGD(lr=0.01, momentum=0.95, decay=5e-4, nesterov=True))
- convertKerasToCaffeNetwork(keras_model, self.name + ".proto", int(batch_size))
- convertKerasToCaffeSolver(keras_model, self.name + ".proto", self.name + "_solver.proto", int(max_iter), int(test_iter), int(test_interval), int(display), lr_policy, weight_decay, regularization_type)
+ keras_model.compile(
+ loss='categorical_crossentropy',
+ optimizer=keras.optimizers.SGD(
+ lr=0.01,
+ momentum=0.95,
+ decay=5e-4,
+ nesterov=True))
+ convertKerasToCaffeNetwork(
+ keras_model,
+ self.name + ".proto",
+ int(batch_size))
+ convertKerasToCaffeSolver(
+ keras_model,
+ self.name + ".proto",
+ self.name + "_solver.proto",
+ int(max_iter),
+ int(test_iter),
+ int(test_interval),
+ int(display),
+ lr_policy,
+ weight_decay,
+ regularization_type)
self.weights = tempfile.mkdtemp() if weights is None else weights
if load_keras_weights:
- convertKerasToSystemMLModel(sparkSession, keras_model, self.weights)
- if labels is not None and (labels.startswith('https:') or labels.startswith('http:')):
+ convertKerasToSystemMLModel(
+ sparkSession, keras_model, self.weights)
+ if labels is not None and (labels.startswith(
+ 'https:') or labels.startswith('http:')):
import urllib
urllib.urlretrieve(labels, os.path.join(weights, 'labels.txt'))
elif labels is not None:
from shutil import copyfile
copyfile(labels, os.path.join(weights, 'labels.txt'))
- super(Keras2DML,self).__init__(sparkSession, self.name + "_solver.proto", input_shape, transferUsingDF)
+ super(
+ Keras2DML,
+ self).__init__(
+ sparkSession,
+ self.name +
+ "_solver.proto",
+ input_shape,
+ transferUsingDF)
if load_keras_weights:
self.load(self.weights)