快速入门Flink(8)——Flink中的流式处理Transformation操作

快速入门Flink(8)——Flink中的流式处理Transformation操作
        上篇博客给大家讲解了DataSource与DataSink本篇文章准备给大家讲解下Stream中的最长用的几种Transformation操作(收藏,收藏,收藏重要事情说三遍)。

一、KeyBy

逻辑上将一个流分成不相交的分区,每个分区包含相同键的元素。在内部,这是通过散 列分区来实现的


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

/**
* @author
* @date 2020/9/23 21:50
* @version 1.0
*/
object StreamKeyBy {
def main(args: Array[String]): Unit = {
//1.构建流处理运行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
//2.使用socket构建数据源
val socketDataSource = env.socketTextStream("node01", 9999)
//3.处理数据
val keyBy = socketDataSource.flatMap(_.split(" ")).map((_, 1)).keyBy(0)
//4.输出
keyBy.print("StreamKeyBy")
//5.任务执行
env.execute("StreamKeyBy")
}
}

二、Connect

用来将两个 dataStream
组装成一个 ConnectedStreams 而且这个 connectedStream 的组成结构就是保留原有的 dataStream 的结构体;这样我们 就可以把不同的数据组装成同一个结构


import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

/**
* @author
* @date 2020/9/23 22:03
* @version 1.0
*/
object StreamConnect {
def main(args: Array[String]): Unit = {
//1.构建批处理运行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
//2.构建2个数据流
val source1 = env.addSource(new MyNoParallelSource).setParallelism(1)
val source2 = env.addSource(new MyNoParallelSource).setParallelism(1)
//3.使用合并流
val connectStream = source1.connect(source2)
val result = connectStream.map(function1 => {
"function1" + function1
}, function2 => {
"function2" + function2
})
//4.输出
result.print()
//5.任务启动
env.execute("StreamConnect")
}

class MyNoParallelSource() extends SourceFunction[Long] {
var count = 1L
var isRunning = true

override def run(sourceContext: SourceFunction.SourceContext[Long]): Unit = {
while (isRunning) {
sourceContext.collect(count)
count += 1
Thread.sleep(1000)
if (count > 5) {
cancel()
}
}
}

override def cancel(): Unit = {
isRunning = false
}
}
}


三、Split 和 select

快速入门Flink(8)——Flink中的流式处理Transformation操作
Split 就是将一个 DataStream 分成两个或者多个 DataStream Select 就是获取分流后对应的数据
需求: 给出数据 1, 2, 3, 4, 5, 6, 7
请使用 split 和 select 把数据中的奇偶数分开,并打印出奇数


import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
* @author
* @date 2020/9/23 22:14
* @version 1.0
*/
object StreamSplit {
def main(args: Array[String]): Unit = {
//1.构建流处理运行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
//2.构建数据集
val source = env.generateSequence(1, 10)
//3.使用split将数据进行切分
val splitStream = source.split(x => {
(x % 2) match {
case 0 => List("偶数")
case 1 => List("奇数")
}
})
//4.获取奇数并打印
val result = splitStream.select("奇数")
result.print()
//5.任务执行
env.execute("StreamSplit")
}
}

原创:https://www.panoramacn.com
源码网提供WordPress源码,帝国CMS源码discuz源码,微信小程序,小说源码,杰奇源码,thinkphp源码,ecshop模板源码,微擎模板源码,dede源码,织梦源码等。

专业搭建小说网站,小说程序,杰奇系列,微信小说系列,app系列小说

快速入门Flink(8)——Flink中的流式处理Transformation操作

免责声明,若由于商用引起版权纠纷,一切责任均由使用者承担。

您必须遵守我们的协议,如果您下载了该资源行为将被视为对《免责声明》全部内容的认可-> 联系客服 投诉资源
www.panoramacn.com资源全部来自互联网收集,仅供用于学习和交流,请勿用于商业用途。如有侵权、不妥之处,请联系站长并出示版权证明以便删除。 敬请谅解! 侵权删帖/违法举报/投稿等事物联系邮箱:2640602276@qq.com
未经允许不得转载:书荒源码源码网每日更新网站源码模板! » 快速入门Flink(8)——Flink中的流式处理Transformation操作
关注我们小说电影免费看
关注我们,获取更多的全网素材资源,有趣有料!
120000+人已关注
分享到:
赞(0) 打赏

评论抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

您的打赏就是我分享的动力!

支付宝扫一扫打赏

微信扫一扫打赏