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

本期导读 【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)

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

在程序中使用 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 (
 ) 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 相关技术问题 欢迎扫码加入钉钉交流群

本文转自网络，原文链接：https://developer.aliyun.com/article/784471
本站部分内容转载于网络，版权归原作者所有，转载之目的在于传播更多优秀技术内容，如有侵权请联系QQ/微信：153890879删除，谢谢！