本期导读 ：【OSS 访问加速】第七讲
主题：Flink 高效 sink 写入 OSS
讲师：重湖，阿里巴巴计算平台事业部 EMR 高级工程师
内容框架：
- 背景介绍
- 功能介绍
- 如何配置
- 如何使用
直播回放链接：（7/8讲）
https://developer.aliyun.com/live/246851
一、背景介绍
Apache Flink 简介
Apache Flink 是新一代大数据计算引擎的代表，以分布式流计算为核心，同时支持批处理。特点：
- 低延时：Flink 流式计算可以做到亚秒甚至毫秒级延时，相比之下 Spark 流计算很难达到秒级
- 高吞吐：Flink 以分布式快照算法实现容错，对吞吐量的影响很小
- 高容错：基于分布式快照算法，Flink 实现了低代价、高效的容错表现，以及 Exactly_Once 语义保证
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 节点状态快照
写入 OSS vs. 写入 亚马逊S3：
- Native 实现：数据写入以 C++ 代码实现，相比 Java 更高效
- 高速读写：多线程读写临时文件，对大于1MB的文件优势尤其明显
- 数据缓存：读写 OSS 实现本地缓存，加速外部访问
OSS 访问加速，JindoFS 提供新支持
二、如何配置
如何配置 JindoFS Flink Connector
环境要求：
- 集群上有开源版本 Flink 软件，版本不低于1.10.1
SDK 配置：
下载所需 SDK 文件：
- jindo-flink-sink-${version}.jar
- jindofs-sdk-${version}.jar
- 下载链接⭐ (Github)：
https://github.com/aliyun/alibabacloud-jindofs/blob/master/docs/jindofs_sdk_download.md
将两个 jar 放置于集群 Flink 目录下 lib 文件夹：
- Flink 根目录通常可由 $FLINK_HOME 环境变量获取
- 集群所有节点均需配置
Java SPI：自动加载资源，无需额外配置
⭐文档链接(Github)：
在程序中使用 JindoFS Flink Connector
确保集群能够访问 OSS Bucket
- 前提：已购买 OSS 产品，OSS 网站链接：
https://www.aliyun.com/product/oss - 确保能够访问 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 相关技术问题，欢迎扫码加入钉钉交流群！
