大佬，有使用flink自定义过percent_rank函数吗？

楼主你好，是的，可以使用阿里云Flink自定义percent_rank函数。

以下是一个示例：

import org.apache.flink.table.functions.AggregateFunction

class PercentRankAccumulator {
  var count: Long = 0
  var sum: Long = 0
}

class PercentRank extends AggregateFunction[Double, PercentRankAccumulator] {
  override def createAccumulator(): PercentRankAccumulator = new PercentRankAccumulator

  def accumulate(acc: PercentRankAccumulator, score: Long): Unit = {
    acc.count += 1
    acc.sum += score
  }

  def getValue(acc: PercentRankAccumulator): Double = {
    if (acc.count == 0) {
      0.0
    } else {
      val rank = 1 - (acc.count - acc.sum.toDouble / acc.count) / acc.count.toDouble
      BigDecimal(rank).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble
    }
  }

  def merge(acc: PercentRankAccumulator, its: java.lang.Iterable[PercentRankAccumulator]): Unit = {
    val iter = its.iterator
    while (iter.hasNext) {
      val other = iter.next()
      acc.count += other.count
      acc.sum += other.sum
    }
  }

  override def getAccumulatorType: TypeInformation[PercentRankAccumulator] = {
    TypeInformation.of(classOf[PercentRankAccumulator])
  }

  override def getResultType: TypeInformation[Double] = {
    TypeInformation.of(classOf[Double])
  }
}

使用该函数：

val tEnv = StreamTableEnvironment.create(env, settings)
tEnv.createTemporarySystemFunction("percent_rank", new PercentRank)

val data = List(
  (1L, 90L),
  (2L, 77L),
  (3L, 88L),
  (4L, 90L),
  (5L, 80L)
)
val table = env.fromCollection(data).toTable(tEnv, 'id, 'score)
tEnv.registerTable("test", table)
val result = tEnv.sqlQuery("SELECT id, score, percent_rank(score) OVER() as rank FROM test")
val ranks = result.toRetractStream[Row].filter(_._1).map(_._2.toString).print()

在此示例中，我们定义了一个PercentRank聚合函数，它计算给定列的百分位数，并使用registerFunction()方法将其注册到TableEnvironment中。然后，我们可以在SQL查询中使用percent_rank(score) OVER()来计算分数的百分位数，并将结果打印到控制台。