diff --git a/dask/array/ma.py b/dask/array/ma.py
--- a/dask/array/ma.py
+++ b/dask/array/ma.py
@@ -2,7 +2,9 @@
 
 import numpy as np
 
+from dask.array import chunk
 from dask.array.core import asanyarray, blockwise, map_blocks
+from dask.array.reductions import reduction
 from dask.array.routines import _average
 from dask.base import normalize_token
 from dask.utils import derived_from
@@ -170,3 +172,21 @@ def set_fill_value(a, fill_value):
 @derived_from(np.ma)
 def average(a, axis=None, weights=None, returned=False):
     return _average(a, axis, weights, returned, is_masked=True)
+
+
+def _chunk_count(x, axis=None, keepdims=None):
+    return np.ma.count(x, axis=axis, keepdims=keepdims)
+
+
+@derived_from(np.ma)
+def count(a, axis=None, keepdims=False, split_every=None):
+    return reduction(
+        a,
+        _chunk_count,
+        chunk.sum,
+        axis=axis,
+        keepdims=keepdims,
+        dtype=np.intp,
+        split_every=split_every,
+        out=None,
+    )
diff --git a/dask/array/utils.py b/dask/array/utils.py
--- a/dask/array/utils.py
+++ b/dask/array/utils.py
@@ -272,6 +272,7 @@ def assert_eq(
     check_meta=True,
     check_chunks=True,
     check_type=True,
+    check_dtype=True,
     equal_nan=True,
     scheduler="sync",
     **kwargs,
@@ -299,7 +300,7 @@ def assert_eq(
         scheduler=scheduler,
     )
 
-    if str(adt) != str(bdt):
+    if check_dtype and str(adt) != str(bdt):
         raise AssertionError(f"a and b have different dtypes: (a: {adt}, b: {bdt})")
 
     try:

diff --git a/dask/array/tests/test_masked.py b/dask/array/tests/test_masked.py
--- a/dask/array/tests/test_masked.py
+++ b/dask/array/tests/test_masked.py
@@ -1,4 +1,5 @@
 import random
+import sys
 from copy import deepcopy
 from itertools import product
 
@@ -393,3 +394,32 @@ def test_arithmetic_results_in_masked():
     sol = x + masked
     assert_eq(res, sol)
     assert isinstance(res.compute(), np.ma.masked_array)
+
+
+def test_count():
+    data = np.arange(120).reshape((12, 10))
+    mask = (data % 3 == 0) | (data % 4 == 0)
+    x = np.ma.masked_where(mask, data)
+    dx = da.from_array(x, chunks=(2, 3))
+
+    for axis in (None, 0, 1):
+        res = da.ma.count(dx, axis=axis)
+        sol = np.ma.count(x, axis=axis)
+        assert_eq(res, sol)
+
+    res = da.ma.count(dx, keepdims=True)
+    sol = np.ma.count(x, keepdims=True)
+    assert_eq(res, sol)
+
+    # Test all masked
+    x = np.ma.masked_all((12, 10))
+    dx = da.from_array(x, chunks=(2, 3))
+    assert_eq(da.ma.count(dx), np.ma.count(x))
+
+    # Test on non-masked array
+    x = np.arange(120).reshape((12, 10))
+    dx = da.from_array(data, chunks=(2, 3))
+    for axis in (None, 0, 1):
+        res = da.ma.count(dx, axis=axis)
+        sol = np.ma.count(x, axis=axis)
+        assert_eq(res, sol, check_dtype=sys.platform != "win32")

Add ma.count to Dask array
Hello,

Dask is missing the [numpy.ma.count](https://numpy.org/doc/stable/reference/generated/numpy.ma.count.html) function, but this would be useful for use in the cf-python library, which is currently undergoing a daskification.

Would you consider putting this in? I have put together a PR that implements it, which will appear shortly.

Many thanks, David



dask/array/tests/test_masked.py::test_count
dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>4] dask/array/tests/test_masked.py::test_basic[<lambda>1] dask/array/tests/test_masked.py::test_mixed_random[<lambda>12] dask/array/tests/test_masked.py::test_basic[<lambda>0] dask/array/tests/test_masked.py::test_basic[<lambda>7] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>11] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>20] dask/array/tests/test_masked.py::test_mixed_random[<lambda>6] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>14] dask/array/tests/test_masked.py::test_mixed_random[<lambda>25] dask/array/tests/test_masked.py::test_reductions_allmasked[prod-f8] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>2] dask/array/tests/test_masked.py::test_reductions_allmasked[sum-i8] dask/array/tests/test_masked.py::test_creation_functions dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>17] dask/array/tests/test_masked.py::test_basic[<lambda>6] dask/array/tests/test_masked.py::test_arg_reductions[argmax] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>18] dask/array/tests/test_masked.py::test_mixed_random[<lambda>18] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>7] dask/array/tests/test_masked.py::test_reductions_allmasked[prod-i8] dask/array/tests/test_masked.py::test_reductions[mean-i8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>16] dask/array/tests/test_masked.py::test_reductions[any-i8] dask/array/tests/test_masked.py::test_tokenize_masked_array dask/array/tests/test_masked.py::test_set_fill_value dask/array/tests/test_masked.py::test_basic[<lambda>14] dask/array/tests/test_masked.py::test_filled dask/array/tests/test_masked.py::test_reductions_allmasked[any-f8] dask/array/tests/test_masked.py::test_reductions[max-i8] dask/array/tests/test_masked.py::test_basic[<lambda>18] dask/array/tests/test_masked.py::test_mixed_random[<lambda>14] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>5] dask/array/tests/test_masked.py::test_mixed_random[<lambda>21] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>19] dask/array/tests/test_masked.py::test_mixed_random[<lambda>22] dask/array/tests/test_masked.py::test_reductions_allmasked[any-i8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>1] dask/array/tests/test_masked.py::test_cumulative dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>23] dask/array/tests/test_masked.py::test_basic[<lambda>16] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>21] dask/array/tests/test_masked.py::test_reductions[any-f8] dask/array/tests/test_masked.py::test_arithmetic_results_in_masked dask/array/tests/test_masked.py::test_mixed_random[<lambda>10] dask/array/tests/test_masked.py::test_mixed_random[<lambda>20] dask/array/tests/test_masked.py::test_reductions_allmasked[max-f8] dask/array/tests/test_masked.py::test_basic[<lambda>21] dask/array/tests/test_masked.py::test_basic[<lambda>23] dask/array/tests/test_masked.py::test_basic[<lambda>25] dask/array/tests/test_masked.py::test_reductions[std-f8] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>1] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>10] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>24] dask/array/tests/test_masked.py::test_arg_reductions[argmin] dask/array/tests/test_masked.py::test_reductions_allmasked[mean-f8] dask/array/tests/test_masked.py::test_basic[<lambda>11] dask/array/tests/test_masked.py::test_reductions[sum-f8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>11] dask/array/tests/test_masked.py::test_reductions_allmasked[std-i8] dask/array/tests/test_masked.py::test_reductions_allmasked[min-f8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>0] dask/array/tests/test_masked.py::test_mixed_random[<lambda>2] dask/array/tests/test_masked.py::test_mixed_random[<lambda>19] dask/array/tests/test_masked.py::test_reductions[all-i8] dask/array/tests/test_masked.py::test_reductions_allmasked[all-i8] dask/array/tests/test_masked.py::test_basic[<lambda>3] dask/array/tests/test_masked.py::test_reductions[prod-i8] dask/array/tests/test_masked.py::test_reductions_allmasked[var-f8] dask/array/tests/test_masked.py::test_reductions_allmasked[sum-f8] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>26] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>3] dask/array/tests/test_masked.py::test_mixed_random[<lambda>17] dask/array/tests/test_masked.py::test_basic[<lambda>8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>13] dask/array/tests/test_masked.py::test_mixed_random[<lambda>8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>3] dask/array/tests/test_masked.py::test_mixed_output_type dask/array/tests/test_masked.py::test_from_array_masked_array dask/array/tests/test_masked.py::test_masked_array dask/array/tests/test_masked.py::test_mixed_random[<lambda>26] dask/array/tests/test_masked.py::test_basic[<lambda>19] dask/array/tests/test_masked.py::test_basic[<lambda>4] dask/array/tests/test_masked.py::test_reductions[all-f8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>24] dask/array/tests/test_masked.py::test_accessors dask/array/tests/test_masked.py::test_reductions[var-f8] dask/array/tests/test_masked.py::test_reductions[mean-f8] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>0] dask/array/tests/test_masked.py::test_reductions_allmasked[min-i8] dask/array/tests/test_masked.py::test_tensordot dask/array/tests/test_masked.py::test_mixed_random[<lambda>5] dask/array/tests/test_masked.py::test_reductions[max-f8] dask/array/tests/test_masked.py::test_reductions_allmasked[std-f8] dask/array/tests/test_masked.py::test_basic[<lambda>12] dask/array/tests/test_masked.py::test_mixed_random[<lambda>7] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>22] dask/array/tests/test_masked.py::test_basic[<lambda>5] dask/array/tests/test_masked.py::test_reductions[var-i8] dask/array/tests/test_masked.py::test_basic[<lambda>24] dask/array/tests/test_masked.py::test_mixed_random[<lambda>15] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>12] dask/array/tests/test_masked.py::test_reductions_allmasked[mean-i8] dask/array/tests/test_masked.py::test_reductions_allmasked[all-f8] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>9] dask/array/tests/test_masked.py::test_reductions[prod-f8] dask/array/tests/test_masked.py::test_basic[<lambda>9] dask/array/tests/test_masked.py::test_basic[<lambda>17] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>13] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>9] dask/array/tests/test_masked.py::test_basic[<lambda>10] dask/array/tests/test_masked.py::test_copy_deepcopy dask/array/tests/test_masked.py::test_basic[<lambda>2] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>15] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>25] dask/array/tests/test_masked.py::test_reductions[min-i8] dask/array/tests/test_masked.py::test_basic[<lambda>15] dask/array/tests/test_masked.py::test_basic[<lambda>20] dask/array/tests/test_masked.py::test_basic[<lambda>13] dask/array/tests/test_masked.py::test_reductions[sum-i8] dask/array/tests/test_masked.py::test_reductions_allmasked[var-i8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>23] dask/array/tests/test_masked.py::test_reductions_allmasked[max-i8] dask/array/tests/test_masked.py::test_basic[<lambda>22] dask/array/tests/test_masked.py::test_average_weights_with_masked_array dask/array/tests/test_masked.py::test_reductions[min-f8] dask/array/tests/test_masked.py::test_mixed_random[<lambda>4] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>6] dask/array/tests/test_masked.py::test_basic[<lambda>26] dask/array/tests/test_masked.py::test_mixed_concatenate[<lambda>16] dask/array/tests/test_masked.py::test_reductions[std-i8]