【数据湖JindoFS+OSS 实操干货36讲】第7/8讲

1 . | E-MapReduce | 对象存储OSS 数据湖 JindoFS + OSS 实操36讲 【OSS 访问加速】 Flink 高效 sink 写入 OSS 演讲人：重湖 阿里巴巴计算平台事业部 EMR 高级工程师 2021.06.02 

2 . 背景介绍 功能介绍 CONTENT 如何配置 如何使用 

3 .Apache Flink 简介 Apache Flink 是新一代大数据计算引擎的代表，以分布式流计算为核心，同时支持批处理。特点： • 低延时：Flink 流式计算可以做到亚秒甚至毫秒级延时，相比之下 Spark 流计算很难达到毫秒级 • 高吞吐：Flink 以分布式快照算法实现容错，对吞吐量的影响很小 • 高容错：基于分布式快照算法，Flink 实现了低代价、高效的容错表现，以及 Exactly_Once 语义保证 

4 .JindoFS Flink Connector 产生背景 阿里云对象存储 Object Storage Service (OSS)： • 海量：无限容量，弹性伸缩 • 安全：12个9的数据安全性，多种加密方式 • 低成本：远低于云磁盘，且有多种存储方式、生命周期管理等节约成本 • 高可靠：服务可用性 99.9% • 已服务于海量用户 Flink 应用广泛： • 流计算领域业内主要解决方案 • Apache 基金会最活跃项目之一 • 未来：流批一体、在线分析 Flink 使用痛点： • 开源 Apache Flink 尚不支持直接写入 OSS • Hadoop OSS SDK 写入性能不一定满足需求 

5 .JindoFS Flink Connector 介绍 整体架构： • 两阶段 Checkpoint (检查点) 机制： • 第一阶段 MPU (MultiPartUpload，分片上传) 写入 OSS • 第二阶段 MPU 提交 • Recoverable Writer 可恢复性写入： • 临时文件以普通文件格式上传 OSS • Sink 节点状态快照 

6 .JindoFS Flink Connector 介绍 写入 OSS vs. 写入亚马逊 S3： • Native 实现：数据写入以 C++ 代码实现，相比 Java 更高效 • 高速读写：多线程读写临时文件，对大于 1MB 的文件优势尤其明显 • 数据缓存：读写 OSS 实现本地缓存，加速外部访问 OSS 访问加速，JindoFS 提供新支持 

7 .如何配置 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)： • https://github.com/aliyun/alibabacloud-jindofs/blob/master/docs/flink/jindofs_sdk_on_flink_for_oss.md 

8 .在程序中使用 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> 

9 .在程序中使用 JindoFS Flink Connector：Java 在程序中开启 Flink Checkpoint • 前提：使用可重发的数据源，如 Kafka • 通过 StreamExecutionEnvironment 对象打开 Checkpoint (示例)： • 建立：StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); • 打开：env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE); 示例程序 • 下文中，outputStream 是一个预先形成的 DataStream<String> 对象，若需写入 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://<user-bucket>/<user-defined-sink-dir>， 最后还需用 env.execute() 语句执行 Flink 作业，env 是已建立的 StreamExecutionEnvironment 对象 • 最后，将 Java 作业打包为 jar 文件，并用 flink run 在集群提交即可 

10 .在程序中使用 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' ) """ 然后将其添加到 StreamTableEnvironment t_env 中即可：t_env.sql_update(sink_ddl) 

11 .在程序中使用 JindoFS Flink Connector：更多配置 用户通过 flink run 提交 java 或 pyflink 程序时，可以额外自定义一些参数，格式： flink run -m yarn-cluster -yD key1=value1 -yD key2=value2 ... 目前支持“熵注入”及“分片上传并行度”两项配置 熵注入 (entropy injection)： • 功能：将写入路径的一段特定字符串匹配出来，用一段随机的字符串进行替换 • 效果：削弱所谓 “片区” (sharding) 效应，提高写入效率 • 配置参数： oss.entropy.key=<user-defined-key> oss.entropy.length=<user-defined-length> 分片上传并行度 • 配置参数：oss.upload.max.concurrent.uploads • 默认值：当前可用的处理器数量 

12 .

13 .

14 .

15 .

16 .

17 .

18 .

19 .

20 .

21 .

22 .