почему возникает ошибка, когда rdd использует sortby с упорядочением

Question

это моя функция

/**
    * 
    *
    * @param spark               spark
    * @param templateInfo        (code,(type_ids,content,lang))
    * @param pushedTemplatedInfo (CODE,PUSH_DATE,PUSHED_CNT)
    * @param templateCycle       
    * @param catTypeId           
    * @param templateCount       
    */
  def getNormalTemplate(spark: SparkSession, templateInfo: RDD[(String, (String, String, String))],
                        pushedTemplatedInfo: RDD[(String, (String, Int))],
                        templateCycle: Int, catTypeId: Int, templateCount: Int) = {
    val templateDate = pushUtil.getNextSomeday(templateCycle)
    println("templateDate:" + templateDate)
    val deleteTemplatedInfo = pushedTemplatedInfo.filter(_._2._1 >= templateDate).map(x => (x._1, x._2._1))
    val brpushedTemplatedMap = spark.sparkContext
      .broadcast(pushedTemplatedInfo.map(x => (x._1, x._2._2)).distinct().collectAsMap())
    val TemplateCodeSelection = templateInfo.filter(x => x._2._1 == catTypeId) 
      .map(x => (x._1, brpushedTemplatedMap.value.getOrElse(x._1, 0))) 
      .reduceByKey((x, y) => math.max(x, y))
      .subtractByKey(deleteTemplatedInfo) 
      .sortBy(x => (x._2, x._1))(Ordering.Tuple2(Ordering.Int,Ordering.String.reverse))

    //(code,(type_ids,content,lang))
    val res = templateInfo.map(x => x._1)
  }

Может кто-нибудь сказать мне, почему, я кодирую порядок следования Как сортироватьсписок в Scala по двум полям?

Answer 1

если вы увидите сигнатуру метода sortBy, вы увидите, что ему нужны 2 параметра Ordering и ClassTag.вам нужно отправить тег класса вашего Tuple

  def sortBy[K](
      f: (T) => K,
      ascending: Boolean = true,
      numPartitions: Int = this.partitions.length)
      (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T] = withScope {
    this.keyBy[K](f)
        .sortByKey(ascending, numPartitions)
        .values
  }

. Вы можете создать тег класса следующим образом:

ClassTag[(Int, String)]((Int, String).getClass)

. Чтобы исправить ситуацию, вы должны вызвать sortBy следующим образом:

.sortBy(x => (x._2, x._1))(Ordering.Tuple2(Ordering.Int,Ordering.String.reverse), ClassTag[(Int, String)]((Int, String).getClass))