class_docstr: A writable datasource that logs node IDs of write tasks, for testing.
class_docstr_tok: ["A","writable","datasource","that","logs","node","IDs","of","write","tasks",",","for","testing","."]
code: def __init__(self):
      @ray.remote
      class DataSink:
          def __init__(self):
              self.rows_written = 0
              self.node_ids = set()
  
          def write(self, node_id: str, block: Block) -> str:
              block = BlockAccessor.for_block(block)
              self.rows_written += block.num_rows()
              self.node_ids.add(node_id)
              return "ok"
  
          def get_rows_written(self):
              return self.rows_written
  
          def get_node_ids(self):
              return self.node_ids
  
      self.data_sink = DataSink.remote()
      self.num_ok = 0
      self.num_failed = 0
  
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 1 to create a new NodeLoggerOutputDatasource object from inherited base classes: Datasource with example: obj = NodeLoggerOutputDatasource()