分享一个Flink checkpoint失败的问题和解决办法原创
接触Flink一段时间了，遇到了一些问题，其中有一个checkpoint失败导致作业重启的问题，遇到了很多次，重启之后一般也能恢复正常，没有太在意，最近2天有同事又频繁遇到，这里记录一下解决方案和分析过程。
我们的flink测试环境有3个节点，部署架构是每个flink节点上部署一个HDFS的DataNode节点，hdfs用于flink的checkpoint和savepoint
现象
看日志是说有个3个datanode活着，文件副本是1，但是写文件失败
There are 3 datanode(s) running and no node(s) are excluded
网上搜了一下这种报错，没有直接的答案，我看了下namenode的日志，没有更多直接的信息
50070 web ui上看一切正常，datanode的剩余空间都还有很多，使用率不到10%
我试了一下往hdfs上put一个文件再get下来，都ok，说明hdfs服务没有问题，datanode也是通的
日志现象1
继续前后翻了一下namenode的日志，注意到有一些warning信息，
这时候怀疑块放置策略有问题
按照日志提示打开相应的的debug开关
修改
etc/hadoop/log4j.properties
找到
log4j.logger.org.apache.hadoop.fs.s3a.S3AFileSystem=WARN
照抄这个格式，在下面添加
log4j.logger.org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicy=DEBUG
log4j.logger.org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor=DEBUG
log4j.logger.org.apache.hadoop.net.NetworkTopology=DEBUG
重启namenode，然后重跑flink作业
日志现象2
这时候看到的问题是机架感知策略无法满足，因为我们没有提供机架映射脚本，默认同一个机架，但是仔细想想跟这个应该也没有关系
因为很多生产环境的hdfs其实都不配置机架映射脚本，并且导致checkpoint失败的问题并不是一直存在，最起码put/get文件都是正常的。
这时候开始考虑看一下hdfs的源码了，根据上面的日志调用栈，先看到BlockPlacementPolicyDefault以及相关的DatanodeDescriptor
这些源码大致的意思是当给一个块选择一个datanode的时候，要对这个datanode进行一些检查，比如看下剩余空间，看下繁忙程度
当我们的问题复现的时候，观察日志会发现一些与此相关的关键信息
这个日志的意思是，存储空间有43G，分配块实际需要100多M,但是scheduled大小就超过43G,因此我们认为正常的datanode，namenode认为它空间不足了
原因
scheduled大小含义是什么呢？ 根据代码可以看到scheduled大小是块大小跟一个计数器做乘法，计数器代表的其实是新建文件块数量计数器，hdfs根据这两个参数评估可能需要的存储空间，相当于给每个datanode预定了一定的空间，预定的空间在文件写入后，计算完真实的占用空间后，还会调整回来。
了解这个原理之后，可以判断的是datanode在一段时间内被预定了太多的空间。
flink的checkpoint机制可以参考这一篇https://www.jianshu.com/p/9c587bd491fc
大致的意思是taskmanager上的很多任务线程都会写hdfs
看了下hdfs的目录结构，有大量的类似uuid命名checkpoint文件，同时每个文件都很小
当我们的作业并发较大时，相应的在hdfs上就会创建更多的checkpoint文件，尽管我们的文件大小只有几K，但是在每一个datanode预定的空间都是128M乘以分配到的文件数量（文件很小，不超过128M），那么43G的空间，最多预定多少文件呢？除一下也就是300多个，三个节点也就是最多900个，我们有多个作业，总并发较大，在预留空间完全释放前，是很容易出现这个问题的。
之前知道hdfs不适合存储小文件，原因是大量小文件会导致inode消耗以及block location这些元数据增长，让namenode内存吃紧，这个例子还表明
当blocksize设置较大，文件大小却远小于blocksize时，大量这种小文件会导致datanode直接"不可用"。
解决办法
块大小不是集群属性，是文件属性，客户端可以设置的，flink这时候每个taskmanager和jobmanager都是hdfs的"客户端"，根据flink文档，我们可以做如下配置
1、在conf/flink-conf.yaml中指定一个hdfs的配置文件路径
fs.hdfs.hadoopconf: /home/xxxx/flink/conf
这里跟flink的配置文件路径选择同一个目录
2、放进去2个配置文件，一个core-site.xml一个是hdfs-site.xml
core-site.xml可以不放，如果checkpoint和savepoint指定了具体的hdfs地址的话，
hdfs-site.xml里加上blockSize配置即可，比如这里我们给它设置为1M
具体块大小如何设置，需要观察自己的作业状态文件大小自己灵活调整。
重启flink集群，提交作业即可，运行时可以观察下hdfs的fsimage大小，注意不要因为块太小，小文件太多导致元数据过大。
小结
我们已经将该问题同步到集群自动化部署脚本中，部署时会专门添加blocksize的配置。
flink这套依赖hdfs的checkpoint方案对于轻量级的流计算场景稍显臃肿，checkpoint的分布式存储不管是直接filesystem还是rocksDB都需要hdfs,其实从checkpoint原理和数据类型考虑，es应该也是不错的选择，遗憾的是社区并没有提供这种方案。
