semvec/pypi-packages huggingface dataset:
dask-tensorflow 0.0.2
Interactions between Dask and Tensorflow
License: BSD

Dask-Tensorflow
===============

.. |Build Status| image:: https://travis-ci.org/mrocklin/dask-tensorflow.svg?branch=master
   :target: https://travis-ci.org/mrocklin/dask-tensorflow

Start TensorFlow clusters from Dask

Example
-------

Given a Dask cluster

.. code-block:: python

   from dask.distributed import Client
   client = Client('scheduler-address:8786')

Get a TensorFlow cluster, specifying groups by name

.. code-block:: python

   from dask_tensorflow import start_tensorflow
   tf_spec, dask_spec = start_tensorflow(client, ps=2, worker=4)

   >>> tf_spec
   {'worker': ['192.168.1.100:2222', '192.168.1.101:2222',
               '192.168.1.102:2222', '192.168.1.103:2222'],
    'ps': ['192.168.1.104:2222', '192.168.1.105:2222']}

This creates a ``tensorflow.train.Server`` on each Dask worker and sets up a
Queue for data transfer on each worker.  These are accessible directly as
``tensorflow_server`` and ``tensorflow_queue`` attributes on the workers.

More Complex Workflow
---------------------

Typically then we set up long running Dask tasks that get these servers and
participate in general TensorFlow compuations.

.. code-block:: python

   from dask.distributed import worker_client

   def ps_function(self):
       with worker_client() as c:
           tf_server = c.worker.tensorflow_server
           tf_server.join()

   ps_tasks = [client.submit(ps_function, workers=worker, pure=False)
               for worker in dask_spec['ps']]

   def worker_function(self):
       with worker_client() as c:
           tf_server = c.worker.tensorflow_server

           # ... use tensorflow as desired ...

   worker_tasks = [client.submit(worker_function, workers=worker, pure=False)
                   for worker in dask_spec['worker']]

One simple and flexible approach is to have these functions block on queues and
feed them data from dask arrays, dataframes, etc.


.. code-block:: python

   def worker_function(self):
       with worker_client() as c:
           tf_server = c.worker.tensorflow_server
           queue = c.worker.tensorflow_queue

           while not stopping_condition():
               batch = queue.get()
               # train with batch

And then dump blocks of numpy and pandas dataframes to these queues

.. code-block:: python

   from distributed.worker_client import get_worker
   def dump_batch(batch):
       worker = get_worker()
       worker.tensorflow_queue.put(batch)


   import dask.dataframe as dd
   df = dd.read_csv('hdfs:///path/to/*.csv')
   # clean up dataframe as necessary
   partitions = df.to_delayed()  # delayed pandas dataframes
   client.map(dump_batch, partitions)