当前位置:网站首页>Flink从入门到真香(7、Sink数据输出-文件)
Flink从入门到真香(7、Sink数据输出-文件)
2020-11-08 12:06:00 【osc_u9mt0sus】
Source 是 Flink 程序的输入,Sink 就是 Flink 程序处理完Source后数据的输出,比如将输出写到文件、sockets、外部系统、或者仅仅是显示(在大数据生态中,很多类似的,比如Flume里也是对应的Source/Channel/Sink),Flink 提供了多种数据输出方式
跟在代码中直接写不同(比如可以在RickMap中open、close、map中直接写)他可以保存一些状态,容错重试机制等等
package com.mafei.sinktest
import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
case class SensorReadingTest3(id: String,timestamp: Long, temperature: Double)
object FileSink {
def main(args: Array[String]): Unit = {
//创建执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
val inputStream= env.readTextFile("/opt/java2020_study/maven/flink1/src/main/resources/sensor.txt")
env.setParallelism(1)
//先转换成样例类类型
val dataStream = inputStream
.map(data =>{
val arr = data.split(",") //按照,分割数据,获取结果
SensorReadingTest3(arr(0), arr(1).toLong,arr(2).toDouble) //生成一个传感器类的数据,参数中传toLong和toDouble是因为默认分割后是字符串类别
})
dataStream.print()
//简单的输出到txt中的方法,已被flink弃用
// dataStream.writeAsText("/opt/java2020_study/maven/flink1/src/main/resources/sink.txt")
//新的输出方式-推荐
dataStream.addSink(
StreamingFileSink.forRowFormat(
new Path("/opt/java2020_study/maven/flink1/src/main/resources/sink2.txt"),
new SimpleStringEncoder[SensorReadingTest3]() //可以在括号中传入编码,默认是udf-8
).build()
)
env.execute("udf test")
}
}
代码结构及最终输出效果:
版权声明
本文为[osc_u9mt0sus]所创,转载请带上原文链接,感谢
https://my.oschina.net/u/4365833/blog/4708100
边栏推荐
- Where is the new target market? What is the anchored product? |Ten questions 2021 Chinese enterprise service
- Top 5 Chinese cloud manufacturers in 2018: Alibaba cloud, Tencent cloud, AWS, telecom, Unicom
- 原创 | 数据资产确权浅议
- 入门级!教你小程序开发不求人(附网盘链接)
- 软件测试培训班出来好找工作么
- Flink的sink实战之一:初探
- 渤海银行百万级罚单不断:李伏安却称治理完善,增速呈下滑趋势
- TiDB 性能竞赛 11.02-11.06
- 为什么 Schnorr 签名被誉为比特币 Segwit 后的最大技术更新
- 你的云服务器可以用来做什么?云服务器有什么用途?
猜你喜欢
阿里出品!视觉计算开发者系列手册(附网盘链接)
Don't look! Full interpretation of Alibaba cloud's original data lake system! (Internet disk link attached)
入门级!教你小程序开发不求人(附网盘链接)
当Kubernetes遇到机密计算,看阿里巴巴如何保护容器内数据的安全!(附网盘链接)
python基本语法 变量
解析Istio访问控制
虚拟机中安装 macOS 11 big sur
2 days, using 4 hours after work to develop a test tool
Istio流量管理--Ingress Gateway
应届生年薪35w+ !倒挂老员工,互联网大厂薪资为何越来越高?
随机推荐
Can you do it with only six characters?
Top 5 Chinese cloud manufacturers in 2018: Alibaba cloud, Tencent cloud, AWS, telecom, Unicom
Is software testing training class easy to find a job
分布式文档存储数据库之MongoDB基础入门
2018中国云厂商TOP5:阿里云、腾讯云、AWS、电信、联通 ...
比Python快20%,就问你兴不兴奋?
一文读懂机器学习“数据中毒”
Share the experience of passing the PMP examination
攻防世界之web新手题
211考研失败后,熬夜了两个月拿下字节offer!【面经分享】
A scheme to improve the memory utilization of flutter
TCP协议如何确保可靠传输
Major changes in Huawei's cloud: Cloud & AI rises to Huawei's fourth largest BG with full fire
个人目前技术栈
Research on WLAN direct connection (peer-to-peer connection or P2P) and cross platform research of IOS
Top 5 Chinese cloud manufacturers in 2018: Alibaba cloud, Tencent cloud, AWS, telecom, Unicom
Personal current technology stack
学习小结(关于深度学习、视觉和学习体会)
当Kubernetes遇到机密计算,看阿里巴巴如何保护容器内数据的安全!(附网盘链接)
Top 5 Chinese cloud manufacturers in 2018: Alibaba cloud, Tencent cloud, AWS, telecom, Unicom