欢迎关注Hadoop、Spark、Flink、Hive、Hbase、Flume等大数据资料分享微信公共账号:iteblog_hadoop
  1. 文章总数:961
  2. 浏览总数:11,488,209
  3. 评论:3873
  4. 分类目录:103 个
  5. 注册用户数:5844
  6. 最后更新:2018年10月17日
过往记忆博客公众号iteblog_hadoop
欢迎关注微信公众号:
iteblog_hadoop
大数据技术博客公众号bigdata_ai
大数据猿:
bigdata_ai

Flink:Scala API函数扩展

  为了保存Scala和Java API之间的一致性,一些允许Scala使用高层次表达式的特性从批处理和流处理的标准API中删除。

  如果你想体验Scala表达式的全部特性,你可以通过隐式转换(implicit conversions)来加强Scala API。

  为了使用这些扩展,在DataSet API中,你仅仅需要引入下面类:

import org.apache.flink.api.scala.extensions._

而在 DataStream API中,你需要引入下面类:

import org.apache.flink.streaming.api.scala.extensions._

当然,你也可以引入一个具体的类,并仅仅使用这个类中的特性。

使用偏函数(partial functions)

  通常情况下,DataSet和DataStream APIs都不支持匿名模式匹配函数来解构tuples, case classes或者collections,如下:

val data: DataSet[(Int, String, Double)] = // [...]
data.map {
  case (id, name, temperature) => // [...]
  // The previous line causes the following compilation error:
  // "The argument types of an anonymous function must be fully known. (SLS 8.5)"
}

值得高兴的是,DataSet和DataStream Scala API都提供了相应的函数扩展,并提供了对匿名模式匹配函数的支持。具体的函数及其使用如下:

DataSet API

MethodOriginalExample
mapWithmap (DataSet)
data.mapWith {
  case (_, value) => value.toString
}
mapPartitionWithmapPartition (DataSet)
     
data.mapPartitionWith {
  case head #:: _ => head
}
flatMapWithflatMap (DataSet)
 
data.flatMapWith {
  case (_, name, visitTimes) => visitTimes.map(name -> _)
}
filterWithfilter (DataSet)
 
data.filterWith {
  case Train(_, isOnTime) => isOnTime
}
reduceWithreduce (DataSet, GroupedDataSet)
 
data.reduceWith {
  case ((_, amount1), (_, amount2)) => amount1 + amount2
}
reduceGroupWithreduceGroup (GroupedDataSet)
 
data.reduceGroupWith {
  case id #:: value #:: _ => id -> value
}
groupingBygroupBy (DataSet)
 
data.groupingBy {
  case (id, _, _) => id
}
sortGroupWithsortGroup (GroupedDataSet)
 
grouped.sortGroupWith(Order.ASCENDING) {
  case House(_, value) => value
}
combineGroupWithcombineGroup (GroupedDataSet)
 
grouped.combineGroupWith {
  case header #:: amounts => amounts.sum
}
projectingapply (JoinDataSet, CrossDataSet)
 
data1.join(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case ((pk, tx), (products, fk)) => tx -> products
  }

data1.cross(data2).projecting {
  case ((a, _), (_, b) => a -> b
}
projectingapply (CoGroupDataSet)
 
data1.coGroup(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case (head1 #:: _, head2 #:: _) => head1 -> head2
  }
}

DataStream API

MethodOriginalExample
mapWithmap (DataStream)
 
data.mapWith {
  case (_, value) => value.toString
}
mapPartitionWithmapPartition (DataStream)
  
data.mapPartitionWith {
  case head #:: _ => head
}
flatMapWithflatMap (DataStream)
 
data.flatMapWith {
  case (_, name, visits) => visits.map(name -> _)
}
filterWithfilter (DataStream)
 
data.filterWith {
  case Train(_, isOnTime) => isOnTime
}
keyingBykeyBy (DataStream)
 
data.keyingBy {
  case (id, _, _) => id
}
mapWithmap (ConnectedDataStream)
 
data.mapWith(
  map1 = case (_, value) => value.toString,
  map2 = case (_, _, value, _) => value + 1
)
flatMapWithflatMap (ConnectedDataStream)
 
data.flatMapWith(
  flatMap1 = case (_, json) => parse(json),
  flatMap2 = case (_, _, json, _) => parse(json)
)
keyingBykeyBy (ConnectedDataStream)
 
data.keyingBy(
  key1 = case (_, timestamp) => timestamp,
  key2 = case (id, _, _) => id
)
reduceWithreduce (KeyedDataStream, WindowedDataStream)
 
data.reduceWith {
  case ((_, sum1), (_, sum2) => sum1 + sum2
}
foldWithfold (KeyedDataStream, WindowedDataStream)
 
data.foldWith(User(bought = 0)) {
  case (User(b), (_, items)) => User(b + items.size)
}
applyWithapply (WindowedDataStream)
 
data.applyWith(0)(
  foldFunction = case (sum, amount) => sum + amount
  windowFunction = case (k, w, sum) => // [...]
)
projectingapply (JoinedDataStream)
 
data1.join(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case ((pk, tx), (products, fk)) => tx -> products
  }

更多关于上面每个函数的语言,可以分别参见DataStreamDataSet API。

如果想精确地使用其中一个函数,在DataSet中可以如下使用:

import org.apache.flink.api.scala.extensions.acceptPartialFunctions

而在DataStream中可以如下使用:

import org.apache.flink.streaming.api.scala.extensions.acceptPartialFunctions

下面的代码片段展示如何在DataSet中使用这些扩展:

object Iteblog {
  import org.apache.flink.api.scala.extensions._
  case class Point(x: Double, y: Double)
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val ds = env.fromElements(Point(1, 2), Point(3, 4), Point(5, 6))
    ds.filterWith {
      case Point(x, _) => x > 1
    }.reduceWith {
      case (Point(x1, y1), (Point(x2, y2))) => Point(x1 + y1, x2 + y2)
    }.mapWith {
      case Point(x, y) => (x, y)
    }.flatMapWith {
      case (x, y) => Seq("x" -> x, "y" -> y)
    }.groupingBy {
      case (id, value) => id
    }
  }
}
本文翻译自:https://ci.apache.org/projects/flink/flink-docs-master/apis/scala_api_extensions.html
本博客文章除特别声明,全部都是原创!
转载本文请加上:转载自过往记忆(https://www.iteblog.com/)
本文链接: 【Flink:Scala API函数扩展】(https://www.iteblog.com/archives/1651.html)
喜欢 (3)
分享 (0)
发表我的评论
取消评论

表情
本博客评论系统带有自动识别垃圾评论功能,请写一些有意义的评论,谢谢!