Additional updates for detecting a distributed scheduler

dask/array/core.py

@@ -3655,13 +3655,18 @@ def to_zarr(
 
     if isinstance(url, zarr.Array):
         z = url
-        if isinstance(z.store, (dict, MutableMapping)) and config.get(
-            "scheduler", ""
-        ) in ("dask.distributed", "distributed"):
-            raise RuntimeError(
-                "Cannot store into in memory Zarr Array using "
-                "the Distributed Scheduler."
-            )
+        if isinstance(z.store, (dict, MutableMapping)):
+            try:
+                from distributed import default_client
+
+                default_client()
+            except (ImportError, ValueError):
+                pass
+            else:
+                raise RuntimeError(
+                    "Cannot store into in memory Zarr Array using "
+                    "the distributed scheduler."
+                )
 
         if region is None:
             arr = arr.rechunk(z.chunks)

dask/dataframe/io/hdf.py

@@ -191,9 +191,18 @@ def to_hdf(
 
     # If user did not specify scheduler and write is sequential default to the
     # sequential scheduler. otherwise let the _get method choose the scheduler
+    try:
+        from distributed import default_client
+
+        default_client()
+        client_available = True
+    except (ImportError, ValueError):
+        client_available = False
+
     if (
         scheduler is None
         and not config.get("scheduler", None)
+        and not client_available
         and single_node
         and single_file
     ):

dask/tests/test_distributed.py

@@ -350,7 +350,7 @@ def test_zarr_in_memory_distributed_err(c):
     a = da.ones((3, 3), chunks=chunks)
     z = zarr.zeros_like(a, chunks=chunks)
 
-    with pytest.raises(RuntimeError):
+    with pytest.raises(RuntimeError, match="distributed scheduler"):
         a.to_zarr(z)