code: def _exactly_sample(rdd, num: int, seed: int):
      from scipy.stats import hypergeom
  
      split_size = rdd.mapPartitionsWithIndex(
          lambda s, it: [(s, sum(1 for _ in it))]
      ).collectAsMap()
      total = sum(split_size.values())
  
      if num > total:
          raise ValueError(
              f"not enough data to sample, own {total} but required {num}"
          )
      # random the size of each split
      sampled_size = {}
      for split, size in split_size.items():
          sampled_size[split] = hypergeom.rvs(M=total, n=size, N=num)
          total = total - size
          num = num - sampled_size[split]
  
      return rdd.mapPartitionsWithIndex(
          _ReservoirSample(
              split_sample_size=sampled_size, seed=seed
          ).func,
          preservesPartitioning=True,
      )
  
imports: import logging
  import struct
  import typing
  from typing import Callable, Iterable, Tuple, Optional, Any
  import pyspark
  from pyspark.rddsampler import RDDSamplerBase
  from fate.arch import URI
  from fate.arch.computing.api import KVTable, ComputingEngine, K, V
  from fate.arch.computing.api._table import _lifted_reduce_to_serdes, get_serdes_by_type
  from ._materialize import materialize, unmaterialize
label_desc: Use image node_id 5 for calling a global function with example usage: _exactly_sample(rdd, num, seed) and returns: rdd