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: Merge two SparkXShards into a single SparkXShards with a database-style join.
  
  :param right: The other SparkXShards to be merged.
  :param how: Type of merge. 'left', 'right', 'outer' or 'inner'. Default is 'inner'.
  :param on: Column name(s) to join on.
  :return: A new merged SparkXShards.
docstr_tok: ["Merge","two","SparkXShards","into","a","single","SparkXShards","with","a","database-style","join",".",":","param","right",":","The","other","SparkXShards","to","be","merged",".",":","param","how",":","Type","of","merge",".","'left","'",",","'right","'",",","'outer","'","or","'inner","'",".","Default","is","'inner","'",".",":","param","on",":","Column","name","(","s",")","to","join","on",".",":","return",":","A","new","merged","SparkXShards","."]
code: def merge(
      self,
      right: "SparkXShards",
      how: str = "inner",
      on: Optional[str] = None,
  ) -> "SparkXShards":
      """
      Merge two SparkXShards into a single SparkXShards with a database-style join.
  
      :param right: The other SparkXShards to be merged.
      :param how: Type of merge. 'left', 'right', 'outer' or 'inner'. Default is 'inner'.
      :param on: Column name(s) to join on.
      :return: A new merged SparkXShards.
      """
      from bigdl.orca.data.utils import spark_df_to_pd_sparkxshards
  
      invalidInputError(
          isinstance(right, SparkXShards),
          "right should be a SparkXShards",
      )
  
      left_df = self.to_spark_df()
      right_df = right.to_spark_df()
      merged = left_df.join(right_df, on=on, how=how)
  
      # count non-empty partitions
      nonEmptyPart = get_spark_context().accumulator(0)
  
      def f(iterator):
          isEmpty = 1
          for x in iterator:
              isEmpty = 0
              break
          nonEmptyPart.add(isEmpty == 0)
  
      merged.rdd.foreachPartition(f)
  
      # repartition evenly according to the index
      if nonEmptyPart.value != merged.rdd.getNumPartitions():
          merged_withIndex_rdd = merged.rdd.zipWithIndex().map(
              lambda p: (p[1], p[0])
          )
          merged = (
              merged_withIndex_rdd.partitionBy(nonEmptyPart.value)
              .map(lambda p: p[1])
              .toDF(merged.schema)
          )
      mergedXShards = spark_df_to_pd_sparkxshards(merged)
      return mergedXShards
  
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 39 for calling the SparkXShards obj's underlying member method code with example usage: obj.merge(right, how, on) and returns: mergedXShards