Flink开发中的问题

1. 流与批处理的区别

• 流处理系统

流处理系统，其节点间数据传输的标准模型是：当一条数据被处理完成后，序列化到缓存中，然后立刻通过网络传输到下一个节点，由下一个节点继续处理。

• 批处理系统

批处理系统，其节点间数据传输的标准模型是：当一条数据被处理完成后，序列化到缓存中，并不会立刻通过网络传输到下一个节点，当缓存写满，就持久化到本地硬盘上，当所有数据都被处理完成后，才开始将处理后的数据通过网络传输到下一个节点。

• flink的流处理和批处理

Flink的执行引擎采用了一种十分灵活的方式，同时支持了这两种数据传输模型：
• Flink以固定的缓存块为单位进行网络数据传输，用户可以通过设置缓存块超时值指定缓存块的传输时机。如果缓存块的超时值为0，则Flink的数据传输方式类似上文所提到流处理系统的标准模型，此时系统可以获得最低的处理延迟
• 如果缓存块的超时值为无限大，则Flink的数据传输方式类似上文所提到批处理系统的标准模型，此时系统可以获得最高的吞吐量
• 同时缓存块的超时值也可以设置为0到无限大之间的任意值。缓存块的超时阈值越小，则Flink流处理执行引擎的数据处理延迟越低，但吞吐量也会降低，反之亦然。通过调整缓存块的超时阈值，用户可根据需求灵活地权衡系统延迟和吞吐量

原文链接：https://blog.csdn.net/shujuelin/article/details/89351157

2. 恢复作业 checkpoint

检查点(checkpoint)的目录是依赖JobID的，每次运行任务都是一个唯一的JobID（好像不能手动设置），所以要找到上一次任务的JobID才能找到检查点。
保存点(savepoint)需要手动触发，而且在指定目录下还生成一个唯一的子目录。

# savepoint
flink run -s /tmp/state.backend/s1/savepoint-17b840-2cfe3bd5bc0c -c flink.HelloWorld target/scala-flink-0.1.jar

# checkpoint
flink run -s /tmp/state.backend/17b840a3d2221b1400ec03f7e3949b17/chk-960 -c flink.HelloWorld target/scala-flink-0.1.jar

检查点和保存点的恢复方法一样的

3. 用流处理批数据,最后一个窗口不计算

• 现象

  用流处理,处理kafka里面的数据时,最后一个窗口会不关闭.导致最后的数据会丢失.

• 原因

  最后一个窗口的水位线还没到 窗口关闭时间.

• 解决方案

  自定义触发器.以机器时钟为准,5秒触发一次.

5. flink 消费kafka的多个topic

1. 传入 List topics , kafka 支持 多个topic.

2. 多个kafka消费,然后用union 连接.

8.Flink state 调优跟注意点

https://blog.csdn.net/qq_31866793/article/details/97272103

9 Flink1.8.0重大更新-Flink中State的自动清除详解

https://blog.csdn.net/u013411339/article/details/90625604

10. 内存溢出

• 现象

  yang gc 时间达到30秒,fullgc 很少发生.

11 linux 内存过多

运行sync将dirty的内容写回硬盘
sync

通过修改proc系统的drop_caches清理free的cache
echo 1 > /proc/sys/vm/drop_caches

13 ask timeout

增加参数

akka.ask.timeout: 100s
web.timeout: 300000

参看:https://www.cnblogs.com/createweb/p/12027737.html

14 Container exited with a non-zero exit code 143

at org.apache.flink.yarn.YarnResourceManager.lambda$onContainersCompleted$0(YarnResourceManager.java:343)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:402)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:195)
	at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
	at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
	at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)

16 Flink 清理过期 Checkpoint 目录的正确姿势

https://www.jianshu.com/p/165a1bf33e4a

17 flink 内存越来越大,越来越慢

将窗口滑动时间由1分钟改为10分钟

18. flink 与 kafka

consumer.setStartFromEarliest();     //从最早的数据开始消费
consumer.setStartFromLatest();       //从最新的数据开始消费
consumer.setStartFromTimestamp(...); //从根据指定的时间戳（ms）处开始消费
consumer.setStartFromGroupOffsets(); //默认从提交的 offset 开始消费

反序列化用 KafkaDeserializationSchema 可以获取到topic的信息

public class ConsumerRecord<K, V> {
    private final String topic;
    private final int partition;
    private final long offset;
    private final long timestamp;
    private final TimestampType timestampType;
    private final long checksum;
    private final int serializedKeySize;
    private final int serializedValueSize;
    private final K key;
    private final V value;
}

21 集群启动

./hadoop-daemon.sh start journalnode
./hadoop-daemon.sh start zkfc
./hadoop-daemon.sh start datanode
./hadoop-daemon.sh start namenode
./yarn-daemon.sh start nodemanager
./yarn-daemon.sh start resourcemanager

./bin/kafka-server-start.sh -daemon ./config/server.properties
./zkServer.sh start

21 flink Reduce、GroupReduce、GroupCombine笔记

reduce

应用于分组DataSet的Reduce转换使用用户定义的reduce函数将每个组减少为单个元素。对于每组输入元素，reduce函数连续地将元素对组合成一个元素，直到每个组只剩下一个元素。

注意，对于ReduceFunction，返回对象的key字段应与输入值匹配。这是因为reduce是可隐式组合（combine）的，并且从combine运算符发出的对象在传递给reduce运算符时再次按key分组。

GroupReduce

应用于分组DataSet的GroupReduce调用用户定义的group-reduce函数转换每个分组。
这与Reduce的区别在于用户定义的函数会立即获得整个组。在组的所有元素上使用Iterable调用该函数，并且可以返回任意数量的结果元素

GroupCombine 分组连接 (少用)

该策略可能不会一次处理所有数据，而是以多个步骤处理

GroupCombine转换是可组合GroupReduceFunction中组合步骤的通用形式。它在某种意义上被概括为允许将输入类型I组合到任意输出类型O.
相反，GroupReduce中的组合步骤仅允许从输入类型I到输出类型I的组合。这是因为reduce步骤中，GroupReduceFunction期望输入类型为I.

在一些应用中，期望在执行附加变换（例如，减小数据大小）之前将DataSet组合成中间格式。这可以通过CombineGroup转换能以非常低的成本实现。

注意：分组数据集上的GroupCombine在内存中使用贪婪策略执行，该策略可能不会一次处理所有数据，而是以多个步骤处理。
它也可以在各个分区上执行，而无需像GroupReduce转换那样进行数据交换。这可能会导致输出的是部分结果，
所以GroupCombine是不能替代GroupReduce操作的，尽管它们的操作内容可能看起来都一样。

22flink 历史服务器

修改历史服务器配置

org.apache.flink.configuration.HistoryServerOptions

historyserver.web.tmpdir  文件地址.

23 Could not deploy Yarn job cluster

新增:

flink-conf.yaml:rest.port: 8082

24 Flink:Could not forward element to next operator

前后时间窗口不一致导致的.

25flink报错org.apache.commons.cli.Option.builder

删除$FLINK_HOME/lib下面的/commons-cli-1.4.jar

26 Flink中的序列化失败问题

声明为@transent

27 Line could not be encoded

Caused by: java.lang.RuntimeException: Line could not be encoded: [49, 56, 49, 49, 90, 77, 119, 66, 54, 48, 54, 71, 48, 53, 55, 50, 48, 49, 53, 48, 56, 48, 53, 49, 56, 52, 52, 48, 56, 109, 49, 106, 124, -26, -84, -94, -24, -65, -114, -28, -67, -65, -25]
	at org.apache.flink.api.java.io.TextValueInputFormat.readRecord(TextValueInputFormat.java:127)
	at org.apache.flink.api.java.io.TextValueInputFormat.readRecord(TextValueInputFormat.java:38)
	at org.apache.flink.api.common.io.DelimitedInputFormat.nextRecord(DelimitedInputFormat.java:520)
	at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:195)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.nio.charset.MalformedInputException: Input length = 1
	at java.nio.charset.CoderResult.throwException(CoderResult.java:281)
	at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:816)
	at org.apache.flink.api.java.io.TextValueInputFormat.readRecord(TextValueInputFormat.java:117)
	... 6 more

解决方案:

Configuration conf = new Configuration();
conf.setBoolean("recursive.file.enumeration", true);
TextValueInputFormat inputFormat = new TextValueInputFormat(new Path(path));
inputFormat.setSkipInvalidLines(true);

28 Embedded metastore is not allowed

解决方案：flink 集成 hive 时 不支持embedded metastore的，配置hive时 需要起一个hive metastore 并在conf文件配置 hive.metastore.uris

29 flink实战--开发中常见的错误与问题

https://blog.csdn.net/aa518189/article/details/103622261

30 Exceeded checkpoint tolerable failure threshold.

重启