code: def test_write_datasource_ray_remote_args(ray_start_cluster):
      ray.shutdown()
      cluster = ray_start_cluster
      cluster.add_node(
          resources={"foo": 100},
          num_cpus=1,
      )
      cluster.add_node(resources={"bar": 100}, num_cpus=1)
  
      ray.init(cluster.address)
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      bar_node_id = ray.get(
          get_node_id.options(resources={"bar": 1}).remote()
      )
  
      output = NodeLoggerOutputDatasource()
      ds = ray.data.range(100, parallelism=10)
      # Pin write tasks to node with "bar" resource.
      ds.write_datasource(
          output, ray_remote_args={"resources": {"bar": 1}}
      )
      assert output.num_ok == 1
      assert output.num_failed == 0
      assert ray.get(output.data_sink.get_rows_written.remote()) == 100
  
      node_ids = ray.get(output.data_sink.get_node_ids.remote())
      assert node_ids == {bar_node_id}
  
imports: import os
  from typing import Iterable, List
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import torchvision
  from fsspec.implementations.http import HTTPFileSystem
  from fsspec.implementations.local import LocalFileSystem
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data.block import Block, BlockAccessor
  from ray.data.datasource import Datasource, DummyOutputDatasource, WriteResult
  from ray.data.datasource.file_meta_provider import _handle_read_os_error
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.data.tests.util import extract_values
  from ray.tests.conftest import *
  from ray.types import ObjectRef
label_desc: Use image node_id 12 for calling a global function with example usage: test_write_datasource_ray_remote_args(ray_start_cluster) and returns: ray