diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index 899175af45f..ba7db88747a 100644
--- a/doc/whats-new.rst
+++ b/doc/whats-new.rst
@@ -30,6 +30,8 @@ Bug fixes
 
 .. _whats-new.0.10.0:
 
+- Properly point DataArray.__dask_scheduler__ to dask.threaded.get
+
 v0.10.0 (20 November 2017)
 --------------------------
 
diff --git a/xarray/core/dataarray.py b/xarray/core/dataarray.py
index 0516b473994..263860dbb74 100644
--- a/xarray/core/dataarray.py
+++ b/xarray/core/dataarray.py
@@ -594,7 +594,7 @@ def __dask_optimize__(self):
 
     @property
     def __dask_scheduler__(self):
-        return self._to_temp_dataset().__dask_optimize__
+        return self._to_temp_dataset().__dask_scheduler__
 
     def __dask_postcompute__(self):
         func, args = self._to_temp_dataset().__dask_postcompute__()
diff --git a/xarray/tests/test_dask.py b/xarray/tests/test_dask.py
index 4e9b0250a6a..0e8a1827026 100644
--- a/xarray/tests/test_dask.py
+++ b/xarray/tests/test_dask.py
@@ -830,3 +830,16 @@ def test_dataarray_with_dask_coords():
     assert not dask.is_dask_collection(array2)
 
     assert all(isinstance(v._variable.data, np.ndarray) for v in array2.coords.values())
+
+
+def test_basic_compute():
+    ds = Dataset({'foo': ('x', range(5)),
+                  'bar': ('x', range(5))}).chunk({'x': 2})
+    for get in [dask.threaded.get,
+                dask.multiprocessing.get,
+                dask.local.get_sync,
+                None]:
+        with dask.set_options(get=get):
+            ds.compute()
+            ds.foo.compute()
+            ds.foo.variable.compute()