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 partitioned using the specified columns.
  This is only applicable for SparkXShards of Pandas DataFrame.
  
  :param cols: specified columns to partition by.
  :param num_partitions: target number of partitions. If not specified,
  the new SparkXShards would keep the current partition number.
  :return: a new SparkXShards.
docstr_tok: ["Return","a","new","SparkXShards","partitioned","using","the","specified","columns",".","This","is","only","applicable","for","SparkXShards","of","Pandas","DataFrame",".",":","param","cols",":","specified","columns","to","partition","by",".",":","param","num_partitions",":","target","number","of","partitions",".","If","not","specified",",","the","new","SparkXShards","would","keep","the","current","partition","number",".",":","return",":","a","new","SparkXShards","."]
code: def partition_by(
      self, cols: str, num_partitions: Optional[int] = None
  ) -> Optional["SparkXShards"]:
      """
  
      Return a new SparkXShards partitioned using the specified columns.
      This is only applicable for SparkXShards of Pandas DataFrame.
  
      :param cols: specified columns to partition by.
      :param num_partitions: target number of partitions. If not specified,
      the new SparkXShards would keep the current partition number.
      :return: a new SparkXShards.
      """
      if self._get_class_name() == "pandas.core.frame.DataFrame":
          import pandas as pd
  
          schema = self.get_schema()
          # if partition by a column
          if isinstance(cols, str):
              if (
                  not isinstance(schema, Dict)
                  or cols not in schema["columns"]
              ):
                  invalidInputError(
                      False,
                      "The partition column is not in the DataFrame",
                  )
              # change data to key value pairs
              rdd = self.rdd.flatMap(
                  lambda df: df.apply(
                      lambda row: (row[cols], row.values.tolist()),
                      axis=1,
                  ).values.tolist()
              )
  
              partition_num = (
                  self.rdd.getNumPartitions()
                  if not num_partitions
                  else num_partitions
              )
              # partition with key
              partitioned_rdd = rdd.partitionBy(partition_num)
          else:
              invalidInputError(
                  False, "Only support partition by a column name"
              )
  
          def merge(iterator):
              data = [value[1] for value in list(iterator)]
              if data:
                  df = pd.DataFrame(
                      data=data, columns=schema["columns"]
                  ).astype(schema["dtypes"])
                  return [df]
              else:
                  # no data in this partition
                  return []
  
          # merge records to df in each partition
          partitioned_shard = SparkXShards(
              partitioned_rdd.mapPartitions(merge)
          )
          self._uncache()
          return partitioned_shard
      else:
          invalidInputError(
              False,
              "Currently only support partition by for XShards"
              " of Pandas DataFrame",
          )
      return None
  
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 16 for calling the SparkXShards obj's underlying member method code with example usage: obj.partition_by(cols, num_partitions) and returns: None, partitioned_shard, list, list