da.store has a compute=False option

Returns dsk and keys

dask/array/core.py

@@ -491,7 +491,7 @@ def compute(*args, **kwargs):
     return results2
 
 
-def store(sources, targets, **kwargs):
+def store(sources, targets, compute=True, **kwargs):
     """ Store dask arrays in array-like objects, overwrite data in target
 
     This stores dask arrays into object that supports numpy-style setitem
@@ -540,7 +540,10 @@ def store(sources, targets, **kwargs):
     updates = [insert_to_ooc(tgt, src) for tgt, src in zip(targets, sources)]
     dsk = merge([src.dask for src in sources] + updates)
     keys = [key for u in updates for key in u]
-    get(dsk, keys, **kwargs)
+    if compute:
+        get(dsk, keys, **kwargs)
+    else:
+        return dsk, keys
 
 
 def blockdims_from_blockshape(shape, blockshape):

dask/array/tests/test_array_core.py

@@ -619,6 +619,9 @@ def test_store():
     assert raises(ValueError, lambda: store(at, at))
     assert raises(ValueError, lambda: store([at, bt], [at, bt]))
 
+    dsk, keys = store([a, b], [at, bt], compute=False)
+    assert isinstance(dsk, dict) and isinstance(keys, list)
+
 
 def test_to_hdf5():
     try: