数据湖实操讲解【OSS 访问加速】第七讲:Flink 高效 sink 写入 OSS-阿里云开发者社区

开发者社区> 阿里云EMR> 正文
登录阅读全文

数据湖实操讲解【OSS 访问加速】第七讲:Flink 高效 sink 写入 OSS

简介: 数据湖 JindoFS+OSS 实操干货 36讲 每周二16点准时直播! 扫文章底部二维码入钉群,线上准时观看~ Github链接: https://github.com/aliyun/alibabacloud-jindofs

本期导读 :【OSS 访问加速】第七讲


主题:Flink 高效 sink 写入 OSS


讲师:重湖,阿里巴巴计算平台事业部 EMR 高级工程师


内容框架:

  • 背景介绍
  • 功能介绍
  • 如何配置
  • 如何使用


直播回放链接:(7/8讲)

https://developer.aliyun.com/live/246851



一、背景介绍


Apache Flink 简介

Apache Flink 是新一代大数据计算引擎的代表,以分布式流计算为核心,同时支持批处理。特点:

  • 低延时:Flink 流式计算可以做到亚秒甚至毫秒级延时,相比之下 Spark 流计算很难达到秒级
  • 高吞吐:Flink 以分布式快照算法实现容错,对吞吐量的影响很小
  • 高容错:基于分布式快照算法,Flink 实现了低代价、高效的容错表现,以及 Exactly_Once 语义保证

image.png


JindoFS Flink Connector 产生背景


阿里云对象存储 Object Storage Service(OSS):

  • 海量:无限容量,弹性伸缩
  • 安全:12个9的数据安全性,多种加密方式
  • 低成本:远低于云磁盘,且有多种存储方式、生命周期管理等节约成本
  • 高可靠:服务可用性 99.9%
  • 已服务于海量用户


Flink 应用广泛:

  • 流计算领域业内主要解决方案
  • Apache 基金会最活跃项目之一
  • 未来:流批一体、在线分析


Flink 使用痛点:

  • 开源 ApacheFlink 尚不支持直接写入 OSS
  • Hadoop OSS SDK 写入性能不一定满足需求


JindoFS Flink Connector 介绍


整体架构:

两阶段 Checkpoint (检查点) 机制:

  • 第一阶段 MPU (MultiPartUpload,分片上传) 写入 OSS
  • 第二阶段 MPU 提交


Recoverable Writer 可恢复性写入:

  • 临时文件以普通文件格式上传 OSS
  • Sink 节点状态快照


image.png

写入 OSS vs.  写入 亚马逊S3:

  • Native 实现:数据写入以 C++ 代码实现,相比 Java 更高效
  • 高速读写:多线程读写临时文件,对大于1MB的文件优势尤其明显
  • 数据缓存:读写 OSS 实现本地缓存,加速外部访问


OSS 访问加速,JindoFS 提供新支持


image.png


二、如何配置


如何配置 JindoFS Flink Connector

环境要求:

  • 集群上有开源版本 Flink 软件,版本不低于1.10.1


SDK 配置:

下载所需 SDK 文件:


将两个 jar 放置于集群 Flink 目录下 lib 文件夹:

  • Flink 根目录通常可由 $FLINK_HOME 环境变量获取
  • 集群所有节点均需配置


Java SPI:自动加载资源,无需额外配置

⭐文档链接(Github):

https://github.com/aliyun/alibabacloud-jindofs/blob/master/docs/flink/jindofs_sdk_on_flink_for_oss.md


在程序中使用 JindoFS Flink Connector

确保集群能够访问 OSS Bucket


使用合适的路径,流式写入OSS Bucket

  • 写入 OSS 须使用 oss:// 前缀路径,类似于:

oss://<user-bucket>/<user-defined-sink-dir>


更多优化!用 JindoFS SDK 加速 OSS 访问,参考

⭐Github:

https://github.com/aliyun/alibabacloud-jindofs/blob/master/docs/jindofs_sdk_vs_hadoop_sdk.md



在程序中使用 JindoFS Flink Connector:Java

在程序中开启 Flink Checkpoint

  • 前提:使用可重发的数据源,如 Kafka


  • 通过 StreamExecutionEnvironment 对象打开 Checkpoint(示例):

建立:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

打开:

env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);


示例程序

  • 下文中,outputStream 是一个预先形成的 DataStream 对象,若需写入 OSS,则可以这样添加 sink:
String outputPath = "oss://<user-bucket>/<user-defined-sink-dir>";
StreamingFileSink<String> sink= StreamingFileSink.forRowFormat(
        new Path(outputPath),
        new SimpleStringEncoder<String>("UTF-8")
).build();
outputStream.addSink(sink);
  • 上述程序指定将 outputStream 中的String 内容写入 OSS 路径 oss:///,最后还需用 env.execute() 语句执行 Flink 作业,env 是已建立的 StreamExecutionEnvironment 对象
  • 最后,将 Java 作业打包为 jar 文件,并用 flink run 在集群提交即可



在程序中使用 JindoFS Flink Connector:Pyflink

与Java 示例类似,在 Pyflink 中使用 JindoFS Flink Connector 与写入 HDFS 等其他介质方式相同,只需:

  • 将写入路径写作合适的 OSS 路径
  • 注意打开 Checkpoint 功能


例如,下列 Python 程序定义了一张位于 OSS 的表:

sink_dest = "oss://<user-bucket>/<user-defined-sink-dir>"
sink_ddl = f""" 
        CREATE TABLE mySink (
                uid INT,
                pid INT
        ) PARTITIONED BY (
                pid
        ) WITH (
                'connector' = 'filesystem',
                'fpath' = '{sink_dest}',
                'format' = 'csv',
                'sink.rolling-policy.file-size' = '2MB',
                'sink.partition-commit.policy.kind' = 'success-file'
        )
"""


然后将其添加到 StreamTableEnvironmentt_env 中即可:t_env.sql_update(sink_ddl)


在程序中使用 JindoFS Flink Connector:更多配置

用户通过 flink run 提交 java 或 pyflink 程序时,可以额外自定义一些参数,格式:

      flink run -m yarn-cluster -yD key1=value1 -yD key2=value2 ...

目前支持“熵注入”及“分片上传并行度”两项配置


熵注入(entropyinjection):

  • 功能:将写入路径的一段特定字符串匹配出来,用一段随机的字符串进行替换
  • 效果:削弱所谓 “片区” (sharding) 效应,提高写入效率
  • 配置参数:

  ​oss.entropy.key=<user-defined-key>

  oss.entropy.length=<user-defined-length>


分片上传并行度

  • 配置参数:oss.upload.max.concurrent.uploads
  • 默认值:当前可用的处理器数量



直接观看第四课(7/8讲)视频回放,获取实例讲解~

https://developer.aliyun.com/live/246851





Github链接:

https://github.com/aliyun/alibabacloud-jindofs


不错过每次直播信息、探讨更多数据湖 JindoFS+OSS 相关技术问题,欢迎扫码加入钉钉交流群!


image.png

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
+ 订阅

阿里巴巴开源大数据技术团队成立阿里云EMR技术圈, 每周推送前沿技术文章,直播分享经典案例、在线答疑,营造纯粹的开源大数据氛围,欢迎加入!加入钉钉群聊阿里云E-MapReduce交流2群,点击进入查看详情 https://qr.dingtalk.com/action/joingroup?code=v1,k1,cNBcqHn4TvG0iHpN3cSc1B86D1831SGMdvGu7PW+sm4=&_dt_no_comment=1&origin=11

官方博客
官网链接