code: def make_ds(
      num_tasks: int,
      num_blocks_per_task: int,
      block_size: int,
      data_format: str,
      num_columns: int,
      ops_spec: List[Dict[str, Any]],
      target_max_block_size: int,
  ) -> ray.data.Dataset:
      ds = ray.data.read_datasource(
          BlockDatasource(),
          num_blocks_per_task=num_blocks_per_task,
          block_size=block_size,
          data_format=data_format,
          num_columns=num_columns,
          parallelism=num_tasks,
      )
      for op_spec in ops_spec:
          op = op_spec.pop("op")
          if op == "flat_map":
              fn = lambda x: [x, x]  # noqa: E731
          else:
              fn = lambda x: x  # noqa: E731
          ds = getattr(ds, op)(fn, **op_spec)
      return ds
  
imports: import argparse
  import json
  import time
  from typing import Dict, List, Any
  import pandas
  import pyarrow
  import numpy
  import ray
  from ray.data.block import BlockMetadata
  from ray.data.context import DataContext, DEFAULT_TARGET_MAX_BLOCK_SIZE
  from ray.data.datasource import Datasource, ReadTask, Reader
label_desc: Use image node_id 1 for calling a global function with example usage: make_ds(num_tasks, num_blocks_per_task, block_size, data_format, num_columns, ops_spec, target_max_block_size) and returns: ds