code: def test_delayed(setup_cluster):
      from dask import delayed
      import numpy as np
  
      def calc_chunk(n: int, i: int):
          rs = np.random.RandomState(i)
          a = rs.uniform(-1, 1, size=(n, 2))
          d = np.linalg.norm(a, axis=1)
          return (d < 1).sum()
  
      def calc_pi(fs, N):
          return sum(fs) * 4 / N
  
      N = 200_000_000
      n = 10_000_000
  
      fs = [delayed(calc_chunk)(n, i) for i in range(N // n)]
      pi = delayed(calc_pi)(fs, N)
  
      dask_res = pi.compute()
      assert dask_res == pi.compute(scheduler=mars_scheduler)
      assert dask_res == convert_dask_collection(pi).execute().fetch()
  
file_path: mars/mars/contrib/dask/tests/test_dask.py
imports: import pytest
  from ....utils import lazy_import
  from ..None import convert_dask_collection, mars_scheduler
label_desc: Use image node_id 1 for calling a global function with example usage: test_delayed(setup_cluster) and returns: unknown, unknown