Hi all, 在使用flink的shuffle功能时,我发现在operator chain中不同的位置进行shuffle,IO速度有非常明显的差距。
比如我的这个例子:
source-->cal--->sort--->SinkConversionToRow--->sink
从hive读数据,计算,排序,转化为外部类型行,写入hive。
当我把shuffle加到cal和sort中间时,
source-->cal-- (rebalance)->sort--->SinkConversionToRow--->sink
shuffle的数据传输IO速度是3G/s
当我把shuffle加到SinkConversionToRow和sink中间时,
source-->cal-- ->sort--->SinkConversionToRow--(rebalance)-->sink
shuffle的数据传输IO速度是0.1G/s
足足差了30倍!
我猜测这是由于SinkConversionToRow将数据转化为了外部格式,外部格式传输速度慢,内部格式传输速度快。
但是为什么差距这么大? 内部格式如何做到传输速度这么快,外部格式又为什么传输速度这么慢?
SinkConversionToRow代码位置: org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink#translateToTransformation*来自志愿者整理的flink
你是用的Filesystem connector读写hdfs的吗? 由于source和sink的并发已经确定了,中间不管哪个阶段进行shuffle,其实对首尾的处理速度应该影响不大。
当我把shuffle加到cal和sort中间时,
source(640并发)-->cal(640并发)-- (rebalance)->sort(64并发)--->SinkConversionToRow(64并发)--->sink(64并发)
shuffle的数据传输IO速度是3G/s,370G文件传输花费2分钟。
当我把shuffle加到SinkConversionToRow和sink中间时,
source(640并发)-->cal(640并发)-- ->sort(640并发)--->SinkConversionToRow(640并发)--(rebalance)-->sink(64并发)
shuffle的数据传输IO速度是0.1G/s,250G文件传输花费40分钟。*来自志愿者整理的flink
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。