diff --git a/docs/.buildinfo b/docs/.buildinfo
index 54c36710..038d092e 100644
--- a/docs/.buildinfo
+++ b/docs/.buildinfo
@@ -1,4 +1,4 @@
# Sphinx build info version 1
# This file hashes the configuration used when building these files. When it is not found, a full rebuild will be done.
-config: c31f3f4c132de601af2c7dd3d80ea76f
+config: 7466c912b353c5a104638dccc44e9814
tags: 645f666f9bcd5a90fca523b33c5a78b7
diff --git a/docs/_modules/index.html b/docs/_modules/index.html
index 821b0c21..ee9b96cc 100644
--- a/docs/_modules/index.html
+++ b/docs/_modules/index.html
@@ -6,7 +6,7 @@
Overview: module code — TensorFlowOnSpark 1.3.0 documentation
+ Overview: module code — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark.TFCluster — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFCluster — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.TFManager — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFManager — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Source code for tensorflowonspark.TFManager
from multiprocessing.managers import BaseManager
from multiprocessing import JoinableQueue
+
[docs] class TFManager ( BaseManager ):
"""Python multiprocessing.Manager for distributed, multi-process communication.""" pass
@@ -58,18 +59,22 @@ Source code for tensorflowonspark.TFManager
qdict = {} # dictionary of queues
kdict = {} # dictionary of key-values
+
def _get ( key ):
return kdict [ key ]
+
def _set ( key , value ):
kdict [ key ] = value
+
def _get_queue ( qname ):
try :
return qdict [ qname ]
except KeyError :
return None
+
[docs] def start ( authkey , queues , mode = 'local' ):
"""Create a new multiprocess.Manager (or return existing one).
@@ -91,12 +96,13 @@
Source code for tensorflowonspark.TFManager
TFManager . register ( 'get' , callable = lambda key : _get ( key ))
TFManager . register ( 'set' , callable = lambda key , value : _set ( key , value ))
if mode == 'remote' :
- mgr = TFManager ( address = ( '' , 0 ), authkey = authkey )
+ mgr = TFManager ( address = ( '' , 0 ), authkey = authkey )
else :
mgr = TFManager ( authkey = authkey )
mgr . start () return mgr
+
[docs] def connect ( address , authkey ):
"""Connect to a multiprocess.Manager.
@@ -113,7 +119,6 @@
Source code for tensorflowonspark.TFManager
m = TFManager ( address , authkey = authkey )
m . connect () return m
-
Navigation
modules |
-
TensorFlowOnSpark 1.3.0 documentation »
+
TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.TFNode — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFNode — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
-
TensorFlowOnSpark 1.3.0 documentation »
+
TensorFlowOnSpark 1.3.1 documentation »Module code »Source code for tensorflowonspark.TFNode
diff --git a/docs/_modules/tensorflowonspark/TFSparkNode.html b/docs/_modules/tensorflowonspark/TFSparkNode.html
index c4ac9c9d..e0859ddd 100644
--- a/docs/_modules/tensorflowonspark/TFSparkNode.html
+++ b/docs/_modules/tensorflowonspark/TFSparkNode.html
@@ -6,7 +6,7 @@
tensorflowonspark.TFSparkNode — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFSparkNode — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.dfutil — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.dfutil — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Source code for tensorflowonspark.dfutil
elif dtype in int64_dtypes :
feature = ( name , tf . train . Feature ( int64_list = tf . train . Int64List ( value = [ row [ name ]])))
elif dtype in bytes_dtypes :
- feature = ( name , tf . train . Feature ( bytes_list = tf . train . BytesList ( value = [ str ( row [ name ])])))
+ if dtype == 'binary' :
+ feature = ( name , tf . train . Feature ( bytes_list = tf . train . BytesList ( value = [ bytes ( row [ name ])])))
+ else :
+ feature = ( name , tf . train . Feature ( bytes_list = tf . train . BytesList ( value = [ str ( row [ name ]) . encode ( 'utf-8' )])))
elif dtype in float_list_dtypes :
feature = ( name , tf . train . Feature ( float_list = tf . train . FloatList ( value = row [ name ])))
elif dtype in int64_list_dtypes :
@@ -219,16 +222,15 @@ Source code for tensorflowonspark.dfutil
"""
# convert from protobuf-like dict to DataFrame-friendly dict
def _get_value ( k , v ):
- # special handling for binary features
- if k in binary_features :
- return bytearray ( v . bytes_list . value [ 0 ])
-
if v . int64_list . value :
result = v . int64_list . value
elif v . float_list . value :
result = v . float_list . value
- else :
- result = v . bytes_list . value
+ else : # string or bytearray
+ if k in binary_features :
+ return bytearray ( v . bytes_list . value [ 0 ])
+ else :
+ return v . bytes_list . value [ 0 ] . decode ( 'utf-8' )
if len ( result ) > 1 : # represent multi-item tensors as python lists
return list ( result )
@@ -278,12 +280,12 @@ Navigation
modules |
-
TensorFlowOnSpark 1.3.0 documentation »
+
TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.gpu_info — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.gpu_info — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
-
TensorFlowOnSpark 1.3.0 documentation »
+
TensorFlowOnSpark 1.3.1 documentation »Module code »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.marker — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.marker — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.pipeline — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.pipeline — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Source code for tensorflowonspark.pipeline
assert local_args . tfrecord_dir , "Please specify --tfrecord_dir to export DataFrame to TFRecord."
if self . getInputMapping ():
# if input mapping provided, filter only required columns before exporting
- dataset = dataset . select ( self . getInputMapping () . keys ())
+ dataset = dataset . select ( list ( self . getInputMapping ()))
logging . info ( "Exporting DataFrame {} as TFRecord to: {} " . format ( dataset . dtypes , local_args . tfrecord_dir ))
dfutil . saveAsTFRecords ( dataset , local_args . tfrecord_dir )
logging . info ( "Done saving" )
@@ -439,7 +439,7 @@ Source code for tensorflowonspark.pipeline
local_args . tensorboard , local_args . input_mode , driver_ps_nodes = local_args . driver_ps_nodes )
if local_args . input_mode == TFCluster . InputMode . SPARK :
# feed data, using a deterministic order for input columns (lexicographic by key)
- input_cols = sorted ( self . getInputMapping () . keys ())
+ input_cols = sorted ( self . getInputMapping ())
cluster . train ( dataset . select ( input_cols ) . rdd , local_args . epochs )
cluster . shutdown ()
@@ -711,12 +711,12 @@ Navigation
modules |
-
TensorFlowOnSpark 1.3.0 documentation »
+
TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.reservation — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.reservation — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
-
TensorFlowOnSpark 1.3.0 documentation »
+
TensorFlowOnSpark 1.3.1 documentation »Module code »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »tensorflowonspark.util — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.util — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Module code »Index — TensorFlowOnSpark 1.3.0 documentation
+ Index — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Welcome to TensorFlowOnSpark’s documentation! — TensorFlowOnSpark 1.3.0 documentation
+ Welcome to TensorFlowOnSpark’s documentation! — TensorFlowOnSpark 1.3.1 documentation
Navigation
next |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Navigation
next |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Python Module Index — TensorFlowOnSpark 1.3.0 documentation
+ Python Module Index — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Search — TensorFlowOnSpark 1.3.0 documentation
+ Search — TensorFlowOnSpark 1.3.1 documentation
Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Navigation
modules |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark.TFCluster module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFCluster module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.TFManager module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFManager module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.TFNode module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFNode module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.TFSparkNode module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.TFSparkNode module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.dfutil module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.dfutil module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.gpu_info module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.gpu_info module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark package — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark package — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark.marker module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.marker module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.pipeline module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.pipeline module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.reservation module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.reservation module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.reservation_client module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.reservation_client module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »tensorflowonspark.util module — TensorFlowOnSpark 1.3.0 documentation
+ tensorflowonspark.util module — TensorFlowOnSpark 1.3.1 documentation
Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »Navigation
previous |
- TensorFlowOnSpark 1.3.0 documentation »TensorFlowOnSpark 1.3.1 documentation »tensorflowonspark package »2.11.8
3.0.3
3.7.0
- 1.4.0
+ 1.8.0
diff --git a/setup.py b/setup.py
index 0a423c72..d0a8cb51 100644
--- a/setup.py
+++ b/setup.py
@@ -3,7 +3,7 @@
setup(
name = 'tensorflowonspark',
packages = ['tensorflowonspark'],
- version = '1.3.0',
+ version = '1.3.1',
description = 'Deep learning with TensorFlow on Apache Spark clusters',
author = 'Yahoo, Inc.',
url = 'https://github.com/yahoo/TensorFlowOnSpark',
diff --git a/tensorflowonspark/TFManager.py b/tensorflowonspark/TFManager.py
index 6cc21154..a83f8c03 100644
--- a/tensorflowonspark/TFManager.py
+++ b/tensorflowonspark/TFManager.py
@@ -10,6 +10,7 @@
from multiprocessing.managers import BaseManager
from multiprocessing import JoinableQueue
+
class TFManager(BaseManager):
"""Python multiprocessing.Manager for distributed, multi-process communication."""
pass
@@ -20,18 +21,22 @@ class TFManager(BaseManager):
qdict = {} # dictionary of queues
kdict = {} # dictionary of key-values
+
def _get(key):
return kdict[key]
+
def _set(key, value):
kdict[key] = value
+
def _get_queue(qname):
try:
return qdict[qname]
except KeyError:
return None
+
def start(authkey, queues, mode='local'):
"""Create a new multiprocess.Manager (or return existing one).
@@ -53,12 +58,13 @@ def start(authkey, queues, mode='local'):
TFManager.register('get', callable=lambda key: _get(key))
TFManager.register('set', callable=lambda key, value: _set(key, value))
if mode == 'remote':
- mgr = TFManager(address=('',0), authkey=authkey)
+ mgr = TFManager(address=('', 0), authkey=authkey)
else:
mgr = TFManager(authkey=authkey)
mgr.start()
return mgr
+
def connect(address, authkey):
"""Connect to a multiprocess.Manager.
@@ -75,4 +81,3 @@ def connect(address, authkey):
m = TFManager(address, authkey=authkey)
m.connect()
return m
-
diff --git a/tensorflowonspark/dfutil.py b/tensorflowonspark/dfutil.py
index 59c47ece..e27cec93 100644
--- a/tensorflowonspark/dfutil.py
+++ b/tensorflowonspark/dfutil.py
@@ -109,7 +109,10 @@ def _toTFFeature(name, dtype, row):
elif dtype in int64_dtypes:
feature = (name, tf.train.Feature(int64_list=tf.train.Int64List(value=[row[name]])))
elif dtype in bytes_dtypes:
- feature = (name, tf.train.Feature(bytes_list=tf.train.BytesList(value=[str(row[name])])))
+ if dtype == 'binary':
+ feature = (name, tf.train.Feature(bytes_list=tf.train.BytesList(value=[bytes(row[name])])))
+ else:
+ feature = (name, tf.train.Feature(bytes_list=tf.train.BytesList(value=[str(row[name]).encode('utf-8')])))
elif dtype in float_list_dtypes:
feature = (name, tf.train.Feature(float_list=tf.train.FloatList(value=row[name])))
elif dtype in int64_list_dtypes:
@@ -181,16 +184,15 @@ def fromTFExample(iter, binary_features=[]):
"""
# convert from protobuf-like dict to DataFrame-friendly dict
def _get_value(k, v):
- # special handling for binary features
- if k in binary_features:
- return bytearray(v.bytes_list.value[0])
-
if v.int64_list.value:
result = v.int64_list.value
elif v.float_list.value:
result = v.float_list.value
- else:
- result = v.bytes_list.value
+ else: # string or bytearray
+ if k in binary_features:
+ return bytearray(v.bytes_list.value[0])
+ else:
+ return v.bytes_list.value[0].decode('utf-8')
if len(result) > 1: # represent multi-item tensors as python lists
return list(result)
diff --git a/tensorflowonspark/pipeline.py b/tensorflowonspark/pipeline.py
index 825af1d4..73d9722a 100755
--- a/tensorflowonspark/pipeline.py
+++ b/tensorflowonspark/pipeline.py
@@ -391,7 +391,7 @@ def _fit(self, dataset):
assert local_args.tfrecord_dir, "Please specify --tfrecord_dir to export DataFrame to TFRecord."
if self.getInputMapping():
# if input mapping provided, filter only required columns before exporting
- dataset = dataset.select(self.getInputMapping().keys())
+ dataset = dataset.select(list(self.getInputMapping()))
logging.info("Exporting DataFrame {} as TFRecord to: {}".format(dataset.dtypes, local_args.tfrecord_dir))
dfutil.saveAsTFRecords(dataset, local_args.tfrecord_dir)
logging.info("Done saving")
@@ -401,7 +401,7 @@ def _fit(self, dataset):
local_args.tensorboard, local_args.input_mode, driver_ps_nodes=local_args.driver_ps_nodes)
if local_args.input_mode == TFCluster.InputMode.SPARK:
# feed data, using a deterministic order for input columns (lexicographic by key)
- input_cols = sorted(self.getInputMapping().keys())
+ input_cols = sorted(self.getInputMapping())
cluster.train(dataset.select(input_cols).rdd, local_args.epochs)
cluster.shutdown()
diff --git a/test/test.py b/test/test.py
index d948e7b7..27a1eed7 100644
--- a/test/test.py
+++ b/test/test.py
@@ -4,6 +4,7 @@
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
+
class SparkTest(unittest.TestCase):
"""Base class for unittests using Spark. Sets up and tears down a cluster per test class"""
diff --git a/test/test_TFCluster.py b/test/test_TFCluster.py
index d51fe271..ac7d7c70 100644
--- a/test/test_TFCluster.py
+++ b/test/test_TFCluster.py
@@ -2,6 +2,7 @@
import test
from tensorflowonspark import TFCluster, TFNode
+
class TFClusterTest(test.SparkTest):
@classmethod
def setUpClass(cls):
@@ -17,12 +18,12 @@ def _map_fun(args, ctx):
import tensorflow as tf
x = tf.constant(args['x'])
y = tf.constant(args['y'])
- sum = tf.add(x,y)
+ sum = tf.add(x, y)
with tf.Session() as sess:
result = sess.run([sum])
assert result[0] == 3
- args = { 'x':1, 'y':2 }
+ args = {'x': 1, 'y': 2}
cluster = TFCluster.run(self.sc, _map_fun, tf_args=args, num_executors=self.num_workers, num_ps=0)
cluster.shutdown()
@@ -41,20 +42,20 @@ def _map_fun(args, ctx):
sq = tf.square(x)
init_op = tf.global_variables_initializer()
sv = tf.train.Supervisor(is_chief=(ctx.task_index == 0),
- init_op=init_op)
+ init_op=init_op)
with sv.managed_session(server.target) as sess:
tf_feed = TFNode.DataFeed(ctx.mgr, False)
while not sv.should_stop() and not tf_feed.should_stop():
- outputs = sess.run([sq], feed_dict={ x: tf_feed.next_batch(10) })
+ outputs = sess.run([sq], feed_dict={x: tf_feed.next_batch(10)})
tf_feed.batch_results(outputs[0])
sv.stop()
- input = [ [x] for x in range(1000) ] # set up input as tensors of shape [1] to match placeholder
+ input = [[x] for x in range(1000)] # set up input as tensors of shape [1] to match placeholder
rdd = self.sc.parallelize(input, 10)
cluster = TFCluster.run(self.sc, _map_fun, tf_args={}, num_executors=self.num_workers, num_ps=0, input_mode=TFCluster.InputMode.SPARK)
rdd_out = cluster.inference(rdd)
rdd_sum = rdd_out.sum()
- self.assertEqual(rdd_sum, sum( [x * x for x in range(1000)] ))
+ self.assertEqual(rdd_sum, sum([x * x for x in range(1000)]))
cluster.shutdown()
diff --git a/test/test_TFNode.py b/test/test_TFNode.py
index a6c71d9b..d6be7d2b 100644
--- a/test/test_TFNode.py
+++ b/test/test_TFNode.py
@@ -3,6 +3,7 @@
import unittest
from tensorflowonspark import TFManager, TFNode
+
class TFNodeTest(unittest.TestCase):
def test_hdfs_path(self):
"""Normalization of absolution & relative string paths depending on filesystem"""
@@ -25,7 +26,7 @@ def test_hdfs_path(self):
def test_datafeed(self):
"""TFNode.DataFeed basic operations"""
- mgr = TFManager.start('abc', ['input', 'output'], 'local')
+ mgr = TFManager.start('abc'.encode('utf-8'), ['input', 'output'], 'local')
# insert 10 numbers followed by an end-of-feed marker
q = mgr.get_queue('input')
diff --git a/test/test_dfutil.py b/test/test_dfutil.py
index dd0e65ef..047c00a7 100644
--- a/test/test_dfutil.py
+++ b/test/test_dfutil.py
@@ -5,6 +5,7 @@
from tensorflowonspark import dfutil
+
class DFUtilTest(test.SparkTest):
@classmethod
def setUpClass(cls):
@@ -30,7 +31,7 @@ def test_dfutils(self):
row1 = ('text string', 1, [2, 3, 4, 5], -1.1, [-2.2, -3.3, -4.4, -5.5], bytearray(b'\xff\xfe\xfd\xfc'))
rdd = self.sc.parallelize([row1])
df1 = self.spark.createDataFrame(rdd, ['a', 'b', 'c', 'd', 'e', 'f'])
- print ("schema: {}".format(df1.schema))
+ print("schema: {}".format(df1.schema))
# save the DataFrame as TFRecords
dfutil.saveAsTFRecords(df1, self.tfrecord_dir)
diff --git a/test/test_pipeline.py b/test/test_pipeline.py
index 7d666f0a..2da38449 100644
--- a/test/test_pipeline.py
+++ b/test/test_pipeline.py
@@ -2,11 +2,13 @@
import os
import shutil
import test
+import time
import unittest
from tensorflowonspark import TFCluster, dfutil
from tensorflowonspark.pipeline import HasBatchSize, HasSteps, Namespace, TFEstimator, TFParams
+
class PipelineTest(test.SparkTest):
@classmethod
def setUpClass(cls):
@@ -14,13 +16,13 @@ def setUpClass(cls):
# create an artificial training dataset of two features with labels computed from known weights
np.random.seed(1234)
- cls.features = np.random.rand(1000,2)
+ cls.features = np.random.rand(1000, 2)
cls.weights = np.array([3.14, 1.618])
cls.labels = np.matmul(cls.features, cls.weights)
# convert to Python types for use with Spark DataFrames
- cls.train_examples = [ (cls.features[i].tolist(), [cls.labels[i].item()]) for i in range(1000) ]
+ cls.train_examples = [(cls.features[i].tolist(), [cls.labels[i].item()]) for i in range(1000)]
# create a simple test dataset
- cls.test_examples = [ ([1.0, 1.0], [0.0]) ]
+ cls.test_examples = [([1.0, 1.0], [0.0])]
# define model_dir and export_dir for tests
cls.model_dir = os.getcwd() + os.sep + "test_model"
@@ -44,13 +46,13 @@ def tearDown(self):
def test_namespace(self):
"""Namespace class initializers"""
# from dictionary
- d = { 'string': 'foo', 'integer': 1, 'float': 3.14, 'array': [1,2,3], 'map': {'a':1, 'b':2} }
+ d = {'string': 'foo', 'integer': 1, 'float': 3.14, 'array': [1, 2, 3], 'map': {'a': 1, 'b': 2}}
n1 = Namespace(d)
self.assertEqual(n1.string, 'foo')
self.assertEqual(n1.integer, 1)
self.assertEqual(n1.float, 3.14)
- self.assertEqual(n1.array, [1,2,3])
- self.assertEqual(n1.map, {'a':1, 'b':2})
+ self.assertEqual(n1.array, [1, 2, 3])
+ self.assertEqual(n1.map, {'a': 1, 'b': 2})
self.assertTrue('string' in n1)
self.assertFalse('extra' in n1)
@@ -59,8 +61,8 @@ def test_namespace(self):
self.assertEqual(n2.string, 'foo')
self.assertEqual(n2.integer, 1)
self.assertEqual(n2.float, 3.14)
- self.assertEqual(n2.array, [1,2,3])
- self.assertEqual(n2.map, {'a':1, 'b':2})
+ self.assertEqual(n2.array, [1, 2, 3])
+ self.assertEqual(n2.map, {'a': 1, 'b': 2})
self.assertTrue('string' in n2)
self.assertFalse('extra' in n2)
@@ -76,10 +78,10 @@ def __init__(self, args):
super(Foo, self).__init__()
self.args = args
- n = Namespace({ 'a': 1, 'b': 2 })
+ n = Namespace({'a': 1, 'b': 2})
f = Foo(n).setBatchSize(10).setSteps(100)
combined_args = f.merge_args_params()
- expected_args = Namespace({ 'a': 1, 'b': 2, 'batch_size': 10, 'steps': 100 })
+ expected_args = Namespace({'a': 1, 'b': 2, 'batch_size': 10, 'steps': 100})
self.assertEqual(combined_args, expected_args)
def test_spark_checkpoint(self):
@@ -91,7 +93,7 @@ def test_spark_checkpoint(self):
# train model
args = {}
estimator = TFEstimator(self.get_function('spark/train'), args) \
- .setInputMapping( { 'col1': 'x', 'col2': 'y_' }) \
+ .setInputMapping({'col1': 'x', 'col2': 'y_'}) \
.setModelDir(self.model_dir) \
.setClusterSize(self.num_workers) \
.setNumPS(1) \
@@ -104,8 +106,8 @@ def test_spark_checkpoint(self):
testDF = self.spark.createDataFrame(self.test_examples, ['c1', 'c2'])
# test model from checkpoint, referencing tensors directly
- model.setInputMapping( { 'c1': 'x' }) \
- .setOutputMapping( { 'y': 'cout' })
+ model.setInputMapping({'c1': 'x'}) \
+ .setOutputMapping({'y': 'cout'})
preds = model.transform(testDF).head() # take first/only result, e.g. [ Row(cout=[4.758000373840332])]
pred = preds.cout[0] # unpack scalar from tensor
self.assertAlmostEqual(pred, np.sum(self.weights), 5)
@@ -119,7 +121,7 @@ def test_spark_saved_model(self):
# train model
args = {}
estimator = TFEstimator(self.get_function('spark/train'), args) \
- .setInputMapping( { 'col1': 'x', 'col2': 'y_' }) \
+ .setInputMapping({'col1': 'x', 'col2': 'y_'}) \
.setExportDir(self.export_dir) \
.setClusterSize(self.num_workers) \
.setNumPS(1) \
@@ -134,8 +136,8 @@ def test_spark_saved_model(self):
# test saved_model using exported signature
model.setTagSet('test_tag') \
.setSignatureDefKey('test_key') \
- .setInputMapping({ 'c1': 'features' }) \
- .setOutputMapping({ 'prediction': 'cout' })
+ .setInputMapping({'c1': 'features'}) \
+ .setOutputMapping({'prediction': 'cout'})
preds = model.transform(testDF).head() # take first/only result
pred = preds.cout[0] # unpack scalar from tensor
expected = np.sum(self.weights)
@@ -144,8 +146,8 @@ def test_spark_saved_model(self):
# test saved_model using custom/direct mapping
model.setTagSet('test_tag') \
.setSignatureDefKey(None) \
- .setInputMapping({ 'c1': 'x'}) \
- .setOutputMapping({ 'y': 'cout1', 'y2': 'cout2' })
+ .setInputMapping({'c1': 'x'}) \
+ .setOutputMapping({'y': 'cout1', 'y2': 'cout2'})
preds = model.transform(testDF).head() # take first/only result
pred = preds.cout1[0] # unpack pred scalar from tensor
squared_pred = preds.cout2[0] # unpack squared pred from tensor
@@ -160,14 +162,14 @@ def test_tf_column_filter(self):
trainDF = self.spark.createDataFrame(self.train_examples, ['col1', 'col2'])
# and add some extra columns
- df = trainDF.withColumn('extra1', trainDF.col1 )
+ df = trainDF.withColumn('extra1', trainDF.col1)
df = df.withColumn('extra2', trainDF.col2)
self.assertEquals(len(df.columns), 4)
# train model
args = {}
estimator = TFEstimator(self.get_function('tf/train'), args, export_fn=self.get_function('tf/export')) \
- .setInputMapping( { 'col1': 'x', 'col2': 'y_' }) \
+ .setInputMapping({'col1': 'x', 'col2': 'y_'}) \
.setInputMode(TFCluster.InputMode.TENSORFLOW) \
.setModelDir(self.model_dir) \
.setExportDir(self.export_dir) \
@@ -175,7 +177,7 @@ def test_tf_column_filter(self):
.setClusterSize(self.num_workers) \
.setNumPS(1) \
.setBatchSize(10)
- model = estimator.fit(df)
+ estimator.fit(df)
self.assertTrue(os.path.isdir(self.model_dir))
self.assertTrue(os.path.isdir(self.tfrecord_dir))
@@ -191,7 +193,7 @@ def test_tf_checkpoint_with_export_fn(self):
# train model
args = {}
estimator = TFEstimator(self.get_function('tf/train'), args, export_fn=self.get_function('tf/export')) \
- .setInputMapping( { 'col1': 'x', 'col2': 'y_' }) \
+ .setInputMapping({'col1': 'x', 'col2': 'y_'}) \
.setInputMode(TFCluster.InputMode.TENSORFLOW) \
.setModelDir(self.model_dir) \
.setExportDir(self.export_dir) \
@@ -201,14 +203,15 @@ def test_tf_checkpoint_with_export_fn(self):
.setBatchSize(10)
model = estimator.fit(trainDF)
self.assertTrue(os.path.isdir(self.model_dir))
+ self.assertTrue(os.path.isdir(self.export_dir))
# create a Spark DataFrame of test examples (features, labels)
testDF = self.spark.createDataFrame(self.test_examples, ['c1', 'c2'])
# test model from checkpoint, referencing tensors directly
model.setTagSet('test_tag') \
- .setInputMapping( { 'c1': 'x' }) \
- .setOutputMapping( { 'y': 'cout1', 'y2': 'cout2' })
+ .setInputMapping({'c1': 'x'}) \
+ .setOutputMapping({'y': 'cout1', 'y2': 'cout2'})
preds = model.transform(testDF).head() # take first/only result, e.g. [ Row(cout=[4.758000373840332])]
pred1, pred2 = preds.cout1[0], preds.cout2[0]
self.assertAlmostEqual(pred1, np.sum(self.weights), 5)
@@ -233,7 +236,7 @@ def _spark_train(args, ctx):
cluster=cluster)):
x = tf.placeholder(tf.float32, [None, 2], name='x')
y_ = tf.placeholder(tf.float32, [None, 1], name='y_')
- w = tf.Variable(tf.truncated_normal([2,1]), name='w')
+ w = tf.Variable(tf.truncated_normal([2, 1]), name='w')
y = tf.matmul(x, w, name='y')
y2 = tf.square(y, name="y2") # extra/optional output for testing multiple output tensors
cost = tf.reduce_mean(tf.square(y_ - y), name='cost')
@@ -242,15 +245,15 @@ def _spark_train(args, ctx):
saver = tf.train.Saver()
sv = tf.train.Supervisor(is_chief=(ctx.task_index == 0),
- init_op=init_op)
+ init_op=init_op)
with sv.managed_session(server.target) as sess:
tf_feed = TFNode.DataFeed(ctx.mgr, input_mapping=args.input_mapping)
while not sv.should_stop() and not tf_feed.should_stop():
batch = tf_feed.next_batch(10)
if args.input_mapping:
if len(batch['x']) > 0:
- feed = { x: batch['x'], y_: batch['y_'] }
- opt = sess.run(optimizer, feed_dict=feed)
+ feed = {x: batch['x'], y_: batch['y_']}
+ sess.run(optimizer, feed_dict=feed)
if sv.is_chief:
if args.model_dir:
@@ -262,8 +265,8 @@ def _spark_train(args, ctx):
# export a saved_model
signatures = {
'test_key': {
- 'inputs': { 'features': x },
- 'outputs': { 'prediction': y },
+ 'inputs': {'features': x},
+ 'outputs': {'prediction': y},
'method_name': 'test'
}
}
@@ -281,9 +284,10 @@ def _tf_train(args, ctx):
tf.reset_default_graph() # reset graph in case we're re-using a Spark python worker
cluster, server = TFNode.start_cluster_server(ctx)
+
def _get_examples(batch_size):
"""Generate test data (mocking a queue_runner of file inputs)"""
- features = tf.random_uniform([batch_size,2]) # (batch_size x 2)
+ features = tf.random_uniform([batch_size, 2]) # (batch_size x 2)
weights = tf.constant([[3.14], [1.618]]) # (2, 1)
labels = tf.matmul(features, weights)
return features, labels
@@ -295,7 +299,7 @@ def _get_examples(batch_size):
worker_device="/job:worker/task:%d" % ctx.task_index,
cluster=cluster)):
x, y_ = _get_examples(10) # no input placeholders, TF code reads (or in this case "generates") input
- w = tf.Variable(tf.truncated_normal([2,1]), name='w')
+ w = tf.Variable(tf.truncated_normal([2, 1]), name='w')
y = tf.matmul(x, w, name='y')
global_step = tf.Variable(0)
@@ -306,7 +310,7 @@ def _get_examples(batch_size):
saver = tf.train.Saver()
sv = tf.train.Supervisor(is_chief=(ctx.task_index == 0),
- init_op=init_op)
+ init_op=init_op)
step = 0
with sv.managed_session(server.target) as sess:
while not sv.should_stop() and step < args.steps:
@@ -320,6 +324,9 @@ def _get_examples(batch_size):
ckpt_name = args.model_dir + "/model.ckpt"
print("Saving checkpoint to: {}".format(ckpt_name))
saver.save(sess, ckpt_name)
+
+ # wait for rest of cluster to connect
+ time.sleep(30)
sv.stop()
def _tf_export(args):
@@ -329,7 +336,7 @@ def _tf_export(args):
tf.reset_default_graph() # reset graph in case we're re-using a Spark python worker
x = tf.placeholder(tf.float32, [None, 2], name='x')
- w = tf.Variable(tf.truncated_normal([2,1]), name='w')
+ w = tf.Variable(tf.truncated_normal([2, 1]), name='w')
y = tf.matmul(x, w, name='y')
y2 = tf.square(y, name="y2") # extra/optional output for testing multiple output tensors
saver = tf.train.Saver()
@@ -343,8 +350,8 @@ def _tf_export(args):
# exported signatures defined in code
signatures = {
'test_key': {
- 'inputs': { 'features': x },
- 'outputs': { 'prediction': y, 'pred2': y2 },
+ 'inputs': {'features': x},
+ 'outputs': {'prediction': y, 'pred2': y2},
'method_name': 'test'
}
}
diff --git a/test/test_reservation.py b/test/test_reservation.py
index 6b7f1858..05bda6fc 100644
--- a/test/test_reservation.py
+++ b/test/test_reservation.py
@@ -3,6 +3,7 @@
from tensorflowonspark.reservation import Reservations, Server, Client
+
class ReservationTest(unittest.TestCase):
def test_reservation_class(self):
"""Test core reservation class, expecting 2 reservations"""
@@ -10,12 +11,12 @@ def test_reservation_class(self):
self.assertFalse(r.done())
# add first reservation
- r.add({'node':1})
+ r.add({'node': 1})
self.assertFalse(r.done())
self.assertEquals(r.remaining(), 1)
# add second reservation
- r.add({'node':2})
+ r.add({'node': 2})
self.assertTrue(r.done())
self.assertEquals(r.remaining(), 0)
@@ -30,7 +31,7 @@ def test_reservation_server(self):
# add first reservation
c = Client(addr)
- resp = c.register({'node':1})
+ resp = c.register({'node': 1})
self.assertEqual(resp, 'OK')
# get list of reservations
@@ -53,8 +54,8 @@ def test_reservation_server_multi(self):
def reserve(num):
c = Client(addr)
- #time.sleep(random.randint(0,5)) # simulate varying start times
- resp = c.register({'node':num})
+ # time.sleep(random.randint(0,5)) # simulate varying start times
+ resp = c.register({'node': num})
self.assertEqual(resp, 'OK')
c.await_reservations()
c.close()