class_docstr: A collection of data which can be pre-processed in parallel on Spark.
class_docstr_tok: ["A","collection","of","data","which","can","be","pre-processed","in","parallel","on","Spark","."]
docstr: Return a new SparkXShards that has exactly num_partitions partitions.
  
  :param num_partitions: target number of partitions
  :return: a new SparkXShards object.
docstr_tok: ["Return","a","new","SparkXShards","that","has","exactly","num_partitions","partitions",".",":","param","num_partitions",":","target","number","of","partitions",":","return",":","a","new","SparkXShards","object","."]
code: def repartition(self, num_partitions: int) -> "SparkXShards":
      """
  
      Return a new SparkXShards that has exactly num_partitions partitions.
  
      :param num_partitions: target number of partitions
      :return: a new SparkXShards object.
      """
      class_name = self._get_class_name()
      if class_name == "pandas.core.frame.DataFrame":
          import pandas as pd
  
          rdd = self.rdd.flatMap(
              lambda df: df.apply(
                  lambda row: (row[0], row.values.tolist()), axis=1
              ).values.tolist()
          ).partitionBy(num_partitions)
  
          schema = self.get_schema()
  
          def merge_rows(iter):
              data = [value[1] for value in list(iter)]
              if data:
                  df = pd.DataFrame(
                      data=data, columns=schema["columns"]
                  ).astype(schema["dtypes"])
                  return [df]
              else:
                  # no data in this partition
                  return iter
  
          repartitioned_shard = self._create(
              rdd.mapPartitions(merge_rows), class_name=class_name
          )
      elif class_name == "builtins.list":
          if num_partitions > self.rdd.getNumPartitions():
              rdd = self.rdd.flatMap(lambda data: data).repartition(
                  num_partitions
              )
  
              repartitioned_shard = self._create(
                  rdd.mapPartitions(lambda iter: [list(iter)]),
                  class_name=class_name,
              )
          else:
              rdd = self.rdd.coalesce(num_partitions)
              from functools import reduce
  
              repartitioned_shard = self._create(
                  rdd.mapPartitions(
                      lambda iter: [
                          reduce(lambda l1, l2: l1 + l2, iter)
                      ]
                  ),  # type:ignore
                  class_name=class_name,
              )  # type:ignore
      elif class_name == "numpy.ndarray":
          elem = self.rdd.first()
          shape = elem.shape
          dtype = elem.dtype
          if len(shape) > 0:
              if num_partitions > self.rdd.getNumPartitions():
                  rdd = self.rdd.flatMap(
                      lambda data: list(data)
                  ).repartition(num_partitions)
  
                  repartitioned_shard = self._create(
                      rdd.mapPartitions(
                          lambda iter: np.stack(
                              [list(iter)], axis=0
                          ).astype(dtype)
                      ),
                      class_name=class_name,
                  )
              else:
                  rdd = self.rdd.coalesce(num_partitions)
                  from functools import reduce
  
                  repartitioned_shard = self._create(
                      rdd.mapPartitions(
                          lambda iter: [
                              np.concatenate(list(iter), axis=0)
                          ]
                      ),
                      class_name=class_name,
                  )
          else:
              repartitioned_shard = self._create(
                  self.rdd.repartition(num_partitions),
                  class_name=class_name,
              )
      elif class_name == "builtins.dict":
          elem = self.rdd.first()
          keys = list(elem.keys())
          dtypes = []
          dict_of_batched_ndarray = True
          # Check if all values are ndarray and shape > 1
          for v in elem.values():
              if v.__class__.__name__ != "ndarray" or len(v.shape) == 0:
                  dict_of_batched_ndarray = False
                  break
              else:
                  dtypes.append(v.dtype)
          if dict_of_batched_ndarray:
  
              def dict_to_unbatched_list(d):
                  values = [list(d[k]) for k in keys]
                  return list(zip(*values))
  
              def to_batched_dict(iter):
                  batch_values = list(zip(*iter))
                  if not batch_values:
                      return []
                  batch_ndarrays = [
                      np.stack(v, axis=0).astype(dtype)
                      for v, dtype in zip(batch_values, dtypes)
                  ]
                  return [dict(zip(keys, batch_ndarrays))]
  
              # If number of records in a partition <= 10, may produce empty partition
              rdd = self.rdd.flatMap(
                  lambda data: dict_to_unbatched_list(data)
              ).repartition(num_partitions)
              repartitioned_shard = self._create(
                  rdd.mapPartitions(lambda iter: to_batched_dict(iter)),
                  class_name=class_name,
              )
          else:
              repartitioned_shard = self._create(
                  self.rdd.repartition(num_partitions),
                  class_name=class_name,
              )
      else:
          repartitioned_shard = self._create(
              self.rdd.repartition(num_partitions),
              class_name=class_name,
          )
      self._uncache()
      return repartitioned_shard
  
imports: from py4j.protocol import Py4JError
  from bigdl.orca.data.utils import *
  from bigdl.orca import OrcaContext
  from bigdl.dllib.utils.common import get_node_and_core_number, get_spark_sql_context, get_spark_context
  from bigdl.dllib.utils import nest
  from bigdl.dllib.utils.log4Error import invalidInputError
  import numpy
  from typing import TYPE_CHECKING, Any
  from typing import Callable, Dict, List, Optional, Tuple, Union
label_desc: Use image node_id 15 for calling the SparkXShards obj's underlying member method code with example usage: obj.repartition(num_partitions) and returns: repartitioned_shard, list, iter, list, list, list