code: def test_data_processsor_distributed(
      fast_dev_run, delete_cached_files, tmpdir, monkeypatch
  ):
      """This test ensures the data optimizer works in a fully distributed settings."""
  
      seed_everything(42)
  
      monkeypatch.setattr(
          data_processor_module.os, "_exit", mock.MagicMock()
      )
  
      _create_dataset_mock = mock.MagicMock()
  
      monkeypatch.setattr(
          data_processor_module, "_create_dataset", _create_dataset_mock
      )
  
      from PIL import Image
  
      input_dir = os.path.join(tmpdir, "dataset")
      os.makedirs(input_dir)
  
      imgs = []
      for i in range(30):
          np_data = np.random.randint(
              255, size=(28, 28), dtype=np.uint32
          )
          img = Image.fromarray(np_data).convert("L")
          imgs.append(img)
          img.save(os.path.join(input_dir, f"{i}.JPEG"))
  
      home_dir = os.path.join(tmpdir, "home")
      monkeypatch.setenv("DATA_OPTIMIZER_HOME_FOLDER", home_dir)
  
      remote_output_dir = os.path.join(tmpdir, "dst")
      os.makedirs(remote_output_dir, exist_ok=True)
  
      cache_dir = os.path.join(tmpdir, "cache_1")
      monkeypatch.setenv("DATA_OPTIMIZER_CACHE_FOLDER", cache_dir)
      data_cache_dir = os.path.join(tmpdir, "data_cache_1")
      monkeypatch.setenv(
          "DATA_OPTIMIZER_DATA_CACHE_FOLDER", data_cache_dir
      )
      monkeypatch.setenv("DATA_OPTIMIZER_NUM_NODES", "2")
      monkeypatch.setenv("DATA_OPTIMIZER_NODE_RANK", "0")
      data_processor = TestDataProcessor(
          input_dir=input_dir,
          num_workers=2,
          delete_cached_files=delete_cached_files,
          fast_dev_run=fast_dev_run,
          output_dir=remote_output_dir,
          num_uploaders=1,
          num_downloaders=1,
      )
      data_processor.run(CustomDataChunkRecipe(chunk_size=2))
  
      fast_dev_run_disabled_chunks_0 = [
          "0-index.json",
          "chunk-0-0.bin",
          "chunk-0-1.bin",
          "chunk-0-2.bin",
          "chunk-0-3.bin",
          "chunk-1-0.bin",
          "chunk-1-1.bin",
          "chunk-1-2.bin",
          "chunk-1-3.bin",
      ]
  
      assert (
          sorted(os.listdir(cache_dir))
          == fast_dev_run_disabled_chunks_0
      )
  
      cache_dir = os.path.join(tmpdir, "cache_2")
      monkeypatch.setenv("DATA_OPTIMIZER_CACHE_FOLDER", cache_dir)
      monkeypatch.setenv("DATA_OPTIMIZER_NUM_NODES", "2")
      monkeypatch.setenv("DATA_OPTIMIZER_NODE_RANK", "1")
      data_processor = TestDataProcessor(
          input_dir=input_dir,
          num_workers=2,
          num_uploaders=1,
          num_downloaders=1,
          delete_cached_files=delete_cached_files,
          fast_dev_run=fast_dev_run,
          output_dir=remote_output_dir,
      )
      data_processor.run(CustomDataChunkRecipe(chunk_size=2))
  
      fast_dev_run_disabled_chunks_1 = [
          "chunk-2-0.bin",
          "chunk-2-1.bin",
          "chunk-2-2.bin",
          "chunk-2-3.bin",
          "chunk-3-0.bin",
          "chunk-3-1.bin",
          "chunk-3-2.bin",
          "chunk-3-3.bin",
          "index.json",
      ]
  
      assert (
          sorted(os.listdir(cache_dir))
          == fast_dev_run_disabled_chunks_1
      )
  
      expected = sorted(
          fast_dev_run_disabled_chunks_0
          + fast_dev_run_disabled_chunks_1
          + ["1-index.json"]
      )
  
      assert sorted(os.listdir(remote_output_dir)) == expected
  
      _create_dataset_mock.assert_called()
  
      assert _create_dataset_mock._mock_mock_calls[0].kwargs == {
          "input_dir": str(input_dir),
          "storage_dir": str(remote_output_dir),
          "dataset_type": "CHUNKED",
          "empty": False,
          "size": 30,
          "num_bytes": 26657,
          "data_format": "jpeg",
          "compression": None,
          "num_chunks": 16,
          "num_bytes_per_chunk": [
              2,
              2,
              2,
              2,
              2,
              2,
              2,
              1,
              2,
              2,
              2,
              1,
              2,
              2,
              2,
              2,
          ],
      }
  
imports: import os
  import random
  import sys
  from typing import Any, List
  from unittest import mock
  import lightning_cloud
  import numpy
  import pytest
  import torch
  from lightning import seed_everything
  from lightning.data.streaming import data_processor
  from lightning.data.streaming import functions
  from lightning.data.streaming.cache import Cache, Dir
  from lightning.data.streaming.data_processor import DataChunkRecipe, DataProcessor, DataTransformRecipe, _download_data_target, _get_item_filesizes, _map_items_to_workers_sequentially, _map_items_to_workers_weighted, _remove_target, _upload_fn, _wait_for_disk_usage_higher_than_threshold, _wait_for_file_to_exist
  from lightning.data.streaming.functions import LambdaDataTransformRecipe, map, optimize
  from lightning_cloud import resolver
  from lightning_utilities.core.imports import RequirementCache
label_desc: Use image node_id 11 for calling a global function with example usage: test_data_processsor_distributed(fast_dev_run, delete_cached_files, tmpdir, monkeypatch) without return types