专栏首页码匠的流水账聊聊flink Table的OrderBy及Limit

聊聊flink Table的OrderBy及Limit

本文主要研究一下flink Table的OrderBy及Limit

实例

Table in = tableEnv.fromDataSet(ds, "a, b, c");
Table result = in.orderBy("a.asc");

Table in = tableEnv.fromDataSet(ds, "a, b, c");

// returns the first 5 records from the sorted result
Table result1 = in.orderBy("a.asc").fetch(5); 

// skips the first 3 records and returns all following records from the sorted result
Table result2 = in.orderBy("a.asc").offset(3);

// skips the first 10 records and returns the next 5 records from the sorted result
Table result3 = in.orderBy("a.asc").offset(10).fetch(5);
  • orderBy方法类似sql的order by;limit则由offset及fetch两个方法构成,类似sql的offset及fetch

Table

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala

class Table(
    private[flink] val tableEnv: TableEnvironment,
    private[flink] val logicalPlan: LogicalNode) {

  //......

  def orderBy(fields: String): Table = {
    val parsedFields = ExpressionParser.parseExpressionList(fields)
    orderBy(parsedFields: _*)
  }

  def orderBy(fields: Expression*): Table = {
    val order: Seq[Ordering] = fields.map {
      case o: Ordering => o
      case e => Asc(e)
    }
    new Table(tableEnv, Sort(order, logicalPlan).validate(tableEnv))
  }

  def offset(offset: Int): Table = {
    new Table(tableEnv, Limit(offset, -1, logicalPlan).validate(tableEnv))
  }

  def fetch(fetch: Int): Table = {
    if (fetch < 0) {
      throw new ValidationException("FETCH count must be equal or larger than 0.")
    }
    this.logicalPlan match {
      case Limit(o, -1, c) =>
        // replace LIMIT without FETCH by LIMIT with FETCH
        new Table(tableEnv, Limit(o, fetch, c).validate(tableEnv))
      case Limit(_, _, _) =>
        throw new ValidationException("FETCH is already defined.")
      case _ =>
        new Table(tableEnv, Limit(0, fetch, logicalPlan).validate(tableEnv))
    }
  }

  //......
}
  • Table的orderBy方法,支持String或Expression类型的参数,其中String类型最终是转为Expression类型;orderBy方法最后使用Sort重新创建了Table;offset及fetch方法,使用Limit重新创建了Table(offset方法创建的Limit其fetch为-1;fetch方法如果之前没有指定offset则创建的Limit的offset为0)

Sort

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/plan/logical/operators.scala

case class Sort(order: Seq[Ordering], child: LogicalNode) extends UnaryNode {
  override def output: Seq[Attribute] = child.output

  override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
    child.construct(relBuilder)
    relBuilder.sort(order.map(_.toRexNode(relBuilder)).asJava)
  }

  override def validate(tableEnv: TableEnvironment): LogicalNode = {
    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
      failValidation(s"Sort on stream tables is currently not supported.")
    }
    super.validate(tableEnv)
  }
}
  • Sort继承了UnaryNode,它的构造器接收Set类型的Ordering,其construct方法使用relBuilder.sort来构建sort条件

Ordering

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/expressions/ordering.scala

abstract class Ordering extends UnaryExpression {
  override private[flink] def validateInput(): ValidationResult = {
    if (!child.isInstanceOf[NamedExpression]) {
      ValidationFailure(s"Sort should only based on field reference")
    } else {
      ValidationSuccess
    }
  }
}

case class Asc(child: Expression) extends Ordering {
  override def toString: String = s"($child).asc"

  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
    child.toRexNode
  }

  override private[flink] def resultType: TypeInformation[_] = child.resultType
}

case class Desc(child: Expression) extends Ordering {
  override def toString: String = s"($child).desc"

  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
    relBuilder.desc(child.toRexNode)
  }

  override private[flink] def resultType: TypeInformation[_] = child.resultType
}
  • Ordering是一个抽象类,它有Asc及Desc两个子类

Limit

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/plan/logical/operators.scala

case class Limit(offset: Int, fetch: Int = -1, child: LogicalNode) extends UnaryNode {
  override def output: Seq[Attribute] = child.output

  override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
    child.construct(relBuilder)
    relBuilder.limit(offset, fetch)
  }

  override def validate(tableEnv: TableEnvironment): LogicalNode = {
    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
      failValidation(s"Limit on stream tables is currently not supported.")
    }
    if (!child.isInstanceOf[Sort]) {
      failValidation(s"Limit operator must be preceded by an OrderBy operator.")
    }
    if (offset < 0) {
      failValidation(s"Offset should be greater than or equal to zero.")
    }
    super.validate(tableEnv)
  }
}
  • Limit继承了UnaryNode,它的构造器接收offset及fetch参数,它的construct方法通过relBuilder.limit来设置offset及fetch

小结

  • Table的orderBy方法类似sql的order by;limit则由offset及fetch两个方法构成,类似sql的offset及fetch
  • Table的orderBy方法,支持String或Expression类型的参数,其中String类型最终是转为Expression类型;orderBy方法最后使用Sort重新创建了Table;offset及fetch方法,使用Limit重新创建了Table(offset方法创建的Limit其fetch为-1;fetch方法如果之前没有指定offset则创建的Limit的offset为0)
  • Sort继承了UnaryNode,它的构造器接收Set类型的Ordering,其construct方法使用relBuilder.sort来构建sort条件;Ordering是一个抽象类,它有Asc及Desc两个子类;Limit继承了UnaryNode,它的构造器接收offset及fetch参数,它的construct方法通过relBuilder.limit来设置offset及fetch

doc

  • OrderBy, Offset & Fetch

本文分享自微信公众号 - 码匠的流水账(geek_luandun)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-01-31

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊flink Table的OrderBy及Limit

    flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala

    codecraft
  • 聊聊WebClient的LoadBalance支持

    spring-cloud-commons-2.0.0.RC1-sources.jar!/org/springframework/cloud/client/loa...

    codecraft
  • 聊聊flink的checkpoint配置

    codecraft
  • (三十七) 初遇python之Decorators装饰器

    各位读者大大们大家好,今天学习python的Decorators装饰器,并记录学习过程欢迎大家一起交流分享。

    亚乐记
  • Hexo+Next指定文章隐藏侧栏

    近期在幕布的活动比较多,想新增一个幕布的作品集页面,所以就依葫芦画瓢,新建一个幕布作品集的页面。

    你好我是森林
  • 眼球轨迹大数据告诉你,看帅哥美女时,男女的关注点分别在哪里。

    当我们漫不经心浏览网站时,如果蹦出一张美女图,您的眼睛会盯向哪里?近日国外有一间调查公司借助眼球追踪技术,分析了消费者的网站浏览重点。其中有几项调查是针对男女生...

    小莹莹
  • 推荐系统中重要却又容易被忽视的问题有哪些

    在个性化推荐系统中的绕不开的经典问题有哪些介绍了推荐下中不可避免都会遇到的问题,这里介绍一些不可忽视的一些内容。

    abs_zero
  • 图像缩放示例

    二维图像的缩放属于仿射变换或者透视变换的范畴,一般可以通过OpenCV的warpAffine()或者warpPerspective()函数实现。 出于兴趣,根据...

    一棹烟波
  • Android高德之旅(15)天气查询废话简介总结

    说到吃饭...呸,说到天气,其实这个并不属于地图或者导航的范畴,为什么高德要提供天气查询呢?两个字:讲究。天气查询可以用来改善app体验的功能,如:在跑步类ap...

    大公爵
  • OpenCV鼠标滑轮事件

    鼠标的滑轮事件实现图像的缩放很方便,具体在回调函数中如下写: 其中scale可以在外部定义为全局变量,通过响应CV_EVENT_MOUSEWHEEL滑轮事件获取...

    一棹烟波

扫码关注云+社区

领取腾讯云代金券