正文
flink优化总结
小程序：扫一扫查出行【扫一扫了解最新限行尾号】
复制小程序
一、高性能Flink SQL优化技巧1、Group Aggregate优化技巧开启MicroBatch或MiniBatch（提升吞吐）
MicroBatch和MiniBatch都是微批处理，只是微批的触发机制略有不同。原理同样是缓存一定的数据后再触发处理，以减少对State的访问，从而提升吞吐并减少数据的输出量。
MiniBatch主要依靠在每个Task上注册的Timer线程来触发微批，需要消耗一定的线程调度性能。MicroBatch是MiniBatch的升级版，主要基于事件消息来触发微批，事件消息会按您指定的时间间隔在源头插入。MicroBatch在元素序列化效率、反压表现、吞吐和延迟性能上都要优于MiniBatch。
适用场景
微批处理通过增加延迟换取高吞吐，如果您有超低延迟的要求，不建议开启微批处理。通常对于聚合的场景，微批处理可以显著的提升系统性能，建议开启。
说明 MicroBatch模式也能解决两级聚合数据抖动问题。
开启方式
MicroBatch和MiniBatch默认关闭，开启方式如下。
# 3.2及以上版本开启Window miniBatch方法（3.2及以上版本默认不开启Window miniBatch）。
sql.exec.mini-batch.window.enabled=true
# 批量输出的间隔时间，在使用microBatch策略时，需要增加该配置，且建议和blink.miniBatch.allowLatencyMs保持一致。
blink.microBatch.allowLatencyMs=5000
# 在使用microBatch时，需要保留以下两个miniBatch配置。
blink.miniBatch.allowLatencyMs=5000
# 防止OOM设置每个批次最多缓存数据的条数。
blink.miniBatch.size=20000
开启LocalGlobal（解决常见数据热点问题）
LocalGlobal优化将原先的Aggregate分成Local+Global两阶段聚合，即MapReduce模型中的Combine+Reduce处理模式。第一阶段在上游节点本地攒一批数据进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到的Accumulator合并（Merge），得到最终的结果（GlobalAgg）。
LocalGlobal本质上能够靠LocalAgg的聚合筛除部分倾斜数据，从而降低GlobalAgg的热点，提升性能。您可以结合下图理解LocalGlobal如何解决数据倾斜的问题。
适用场景
LocalGlobal适用于提升如SUM、COUNT、MAX、MIN和AVG等普通聚合的性能，以及解决这些场景下的数据热点问题。
说明 开启LocalGlobal需要UDAF实现Merge方法。
开启方式
实时计算2.0版本开始，LocalGlobal是默认开启的，参数是blink.localAgg.enabled=true ，但是需要在microbatch或minibatch开启的前提下才能生效。
判断是否生效
观察最终生成的拓扑图的节点名字中是否包含GlobalGroupAggregate或LocalGroupAggregate。
开启PartialFinal（解决COUNT DISTINCT热点问题）
LocalGlobal优化针对普通聚合（例如SUM、COUNT、MAX、MIN和AVG）有较好的效果，对于COUNT DISTINCT收效不明显，因为COUNT DISTINCT在Local聚合时，对于DISTINCT KEY的去重率不高，导致在Global节点仍然存在热点。
之前，为了解决COUNT DISTINCT的热点问题，通常需要手动改写为两层聚合（增加按Distinct Key取模的打散层）。自2.2.0版本开始，实时计算提供了COUNT DISTINCT自动打散，即PartialFinal优化，您无需自行改写为两层聚合。PartialFinal和LocalGlobal的原理对比参见下图。
适用场景
使用COUNT DISTINCT，但无法满足聚合节点性能要求。
说明
- 不能在包含UDAF的Flink SQL中使用PartialFinal优化方法。
- 数据量不大的情况下，不建议使用PartialFinal优化方法。PartialFinal优化会自动打散成两层聚合，引入额外的网络Shuffle，在数据量不大的情况下，浪费资源。
开启方式
默认不开启，使用参数显式开启blink.partialAgg.enabled=true。
判断是否生效
观察最终生成的拓扑图的节点名中是否包含Expand节点，或者原来一层的聚合变成了两层的聚合。
改写为AGG WITH FILTER语法（提升大量COUNT DISTINCT场景性能）
说明 仅实时计算2.2.2及以上版本支持AGG WITH FILTER语法。
统计作业需要计算各种维度的UV，例如全网UV、来自手机客户端的UV、来自PC的UV等等。建议使用标准的AGG WITH FILTER语法来代替CASE WHEN实现多维度统计的功能。实时计算目前的SQL优化器能分析出Filter参数，从而同一个字段上计算不同条件下的COUNT DISTINCT能共享State，减少对State的读写操作。性能测试中，使用AGG WITH FILTER语法来代替CASE WHEN能够使性能提升1倍。
适用场景
建议您将AGG WITH CASE WHEN的语法都替换成AGG WITH FILTER的语法，尤其是对同一个字段上计算不同条件下的COUNT DISTINCT结果，性能提升很大。
原始写法
COUNT(distinct visitor_id) as UV1 , COUNT(distinct case when is_wireless='y' then visitor_id else null end) as UV2
优化写法
COUNT(distinct visitor_id) as UV1 , COUNT(distinct visitor_id) filter (where is_wireless='y') as UV2
2、TopN优化技巧TopN算法
当TopN的输入是非更新流（例如Source），TopN只有一种算法AppendRank。当TopN的输入是更新流时（例如经过了AGG/JOIN计算），TopN有3种算法，性能从高到低分别是：UpdateFastRank 、 UnaryUpdateRank和RetractRank。算法名字会显示在拓扑图的节点名字上。
UpdateFastRank ：最优算法。
需要具备2个条件：
输入流有PK信息。
排序字段的更新是单调的，且单调方向与排序方向相反。例如，ORDER BY COUNT/COUNT_DISTINCT/SUM（正数）DESC（仅实时计算2.2.2及以上版本支持）。
如果您要获取到优化Plan，需要告诉系统total_fee
insert
into print_test
SELECT
cate_id,
seller_id,
stat_date,
pay_ord_amt --不输出rownum字段，能减小结果表的输出量。
FROM (
SELECT
*,
ROW_NUMBER () OVER (
PARTITION BY cate_id,
stat_date --注意要有时间字段，否则state过期会导致数据错乱。
ORDER
BY pay_ord_amt DESC
) as rownum --根据上游sum结果排序。
FROM (
SELECT
cate_id,
seller_id,
stat_date,
--重点。声明Sum的参数都是正数，所以Sum的结果是单调递增的，因此TopN能使用优化算法，只获取前100个数据。
sum (total_fee) filter (
where
total_fee >= 0
) as pay_ord_amt
FROM
random_test
WHERE
total_fee >= 0
GROUP
BY cate_name,
seller_id,
stat_date
) a
WHERE
rownum <= 100
);
UnaryUpdateRank：仅次于UpdateFastRank的算法。需要具备1个条件：输入流中存在PK信息。例如，ORDER BY AVG。
RetractRank：普通算法，性能最差，不建议在生产环境使用该算法。请检查输入流是否存在PK信息，如果存在，则可进行UnaryUpdateRank或UpdateFastRank优化。
TopN优化方法
无排名优化
TopN的输出结果无需要显示rownum值，仅需在最终前端显式时进行1次排序，极大地减少输入结果表的数据量。无排名优化方法详情请参见TopN语句。
增加TopN的Cache大小
TopN为了提升性能有一个State Cache层，Cache层能提升对State的访问效率。TopN的Cache命中率的计算公式为。
cache_hit = cache_size*parallelism/top_n/partition_key_num
例如，Top100配置缓存10000条，并发50，当您的PatitionBy的key维度较大时，例如10万级别时，Cache命中率只有10000*50/100/100000=5%，命中率会很低，导致大量的请求都会击中State（磁盘），性能会大幅下降。因此当PartitionKey维度特别大时，可以适当加大TopN的CacheS ize，相对应的也建议适当加大TopN节点的Heap Memory（请参见
手动配置调优
）。
##默认10000条，调整TopN cahce到20万，那么理论命中率能达200000*50/100/100000 = 100%。
blink.topn.cache.size=200000
PartitionBy的字段中要有时间类字段
例如每天的排名，要带上Day字段。否则TopN的结果到最后会由于State ttl有错乱。
3、高效去重方案
开启MicroBatch或MiniBatch（提升吞吐）
MicroBatch和MiniBatch都是微批处理，只是微批的触发机制略有不同。原理同样是缓存一定的数据后再触发处理，以减少对State的访问，从而提升吞吐并减少数据的输出量。
MiniBatch主要依靠在每个Task上注册的Timer线程来触发微批，需要消耗一定的线程调度性能。MicroBatch是MiniBatch的升级版，主要基于事件消息来触发微批，事件消息会按您指定的时间间隔在源头插入。MicroBatch在元素序列化效率、反压表现、吞吐和延迟性能上都要优于MiniBatch。
适用场景
微批处理通过增加延迟换取高吞吐，如果您有超低延迟的要求，不建议开启微批处理。通常对于聚合的场景，微批处理可以显著的提升系统性能，建议开启。
说明 MicroBatch模式也能解决两级聚合数据抖动问题。
开启方式
MicroBatch和MiniBatch默认关闭，开启方式如下。
# 3.2及以上版本开启Window miniBatch方法（3.2及以上版本默认不开启Window miniBatch）。
sql.exec.mini-batch.window.enabled=true
# 批量输出的间隔时间，在使用microBatch策略时，需要增加该配置，且建议和blink.miniBatch.allowLatencyMs保持一致。
blink.microBatch.allowLatencyMs=5000
# 在使用microBatch时，需要保留以下两个miniBatch配置。
blink.miniBatch.allowLatencyMs=5000
# 防止OOM设置每个批次最多缓存数据的条数。
blink.miniBatch.size=20000
开启LocalGlobal（解决常见数据热点问题）
LocalGlobal优化将原先的Aggregate分成Local+Global两阶段聚合，即MapReduce模型中的Combine+Reduce处理模式。第一阶段在上游节点本地攒一批数据进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到的Accumulator合并（Merge），得到最终的结果（GlobalAgg）。
LocalGlobal本质上能够靠LocalAgg的聚合筛除部分倾斜数据，从而降低GlobalAgg的热点，提升性能。您可以结合下图理解LocalGlobal如何解决数据倾斜的问题。
适用场景
LocalGlobal适用于提升如SUM、COUNT、MAX、MIN和AVG等普通聚合的性能，以及解决这些场景下的数据热点问题。
说明 开启LocalGlobal需要UDAF实现
Merge方法。开启方式
实时计算2.0版本开始，LocalGlobal是默认开启的，参数是blink.localAgg.enabled=true ，但是需要在microbatch或minibatch开启的前提下才能生效。
判断是否生效
观察最终生成的拓扑图的节点名字中是否包含GlobalGroupAggregate或LocalGroupAggregate。
开启PartialFinal（解决COUNT DISTINCT热点问题）
LocalGlobal优化针对普通聚合（例如SUM、COUNT、MAX、MIN和AVG）有较好的效果，对于COUNT DISTINCT收效不明显，因为COUNT DISTINCT在Local聚合时，对于DISTINCT KEY的去重率不高，导致在Global节点仍然存在热点。
之前，为了解决COUNT DISTINCT的热点问题，通常需要手动改写为两层聚合（增加按Distinct Key取模的打散层）。自
2.2.0版本开始，实时计算提供了COUNT DISTINCT自动打散，即PartialFinal优化，您无需自行改写为两层聚合。PartialFinal和LocalGlobal的原理对比参见下图。适用场景
使用COUNT DISTINCT，但无法满足聚合节点性能要求。
说明
- 不能在包含UDAF的Flink SQL中使用PartialFinal优化方法。
- 数据量不大的情况下，不建议使用PartialFinal优化方法。PartialFinal优化会自动打散成两层聚合，引入额外的网络Shuffle，在数据量不大的情况下，浪费资源。
开启方式
默认不开启，使用参数显式开启
blink.partialAgg.enabled=true。判断是否生效
观察最终生成的拓扑图的节点名中是否包含Expand节点，或者原来一层的聚合变成了两层的聚合。
改写为AGG WITH FILTER语法（提升大量COUNT DISTINCT场景性能）
说明 仅实时计算2.2.2及以上版本支持AGG WITH FILTER语法。
统计作业需要计算各种维度的UV，例如全网UV、来自手机客户端的UV、来自PC的UV等等。建议使用标准的AGG WITH FILTER语法来代替CASE WHEN实现多维度统计的功能。实时计算目前的SQL优化器能分析出Filter参数，从而同一个字段上计算不同条件下的COUNT DISTINCT能共享State，减少对State的读写操作。性能测试中，使用AGG WITH FILTER语法来代替CASE WHEN能够使性能提升1倍。
适用场景
建议您将AGG WITH CASE WHEN的语法都替换成AGG WITH FILTER的语法，尤其是对同一个字段上计算不同条件下的COUNT DISTINCT结果，性能提升很大。
原始写法
COUNT(distinct visitor_id) as UV1 , COUNT(distinct case when is_wireless='y' then visitor_id else null end) as UV2优化写法
COUNT(distinct visitor_id) as UV1 , COUNT(distinct visitor_id) filter (where is_wireless='y') as UV2
2、TopN优化技巧TopN算法
当TopN的输入是非更新流（例如Source），TopN只有一种算法AppendRank。当TopN的输入是更新流时（例如经过了AGG/JOIN计算），TopN有3种算法，性能从高到低分别是：UpdateFastRank 、 UnaryUpdateRank和RetractRank。算法名字会显示在拓扑图的节点名字上。
UpdateFastRank ：最优算法。
需要具备2个条件：
输入流有PK信息。
排序字段的更新是单调的，且单调方向与排序方向相反。例如，ORDER BY COUNT/COUNT_DISTINCT/SUM（正数）DESC（仅实时计算2.2.2及以上版本支持）。
如果您要获取到优化Plan，需要告诉系统total_fee
insert
into print_test
SELECT
cate_id,
seller_id,
stat_date,
pay_ord_amt --不输出rownum字段，能减小结果表的输出量。
FROM (
SELECT
*,
ROW_NUMBER () OVER (
PARTITION BY cate_id,
stat_date --注意要有时间字段，否则state过期会导致数据错乱。
ORDER
BY pay_ord_amt DESC
) as rownum --根据上游sum结果排序。
FROM (
SELECT
cate_id,
seller_id,
stat_date,
--重点。声明Sum的参数都是正数，所以Sum的结果是单调递增的，因此TopN能使用优化算法，只获取前100个数据。
sum (total_fee) filter (
where
total_fee >= 0
) as pay_ord_amt
FROM
random_test
WHERE
total_fee >= 0
GROUP
BY cate_name,
seller_id,
stat_date
) a
WHERE
rownum <= 100
);
UnaryUpdateRank：仅次于UpdateFastRank的算法。需要具备1个条件：输入流中存在PK信息。例如，ORDER BY AVG。
RetractRank：普通算法，性能最差，不建议在生产环境使用该算法。请检查输入流是否存在PK信息，如果存在，则可进行UnaryUpdateRank或UpdateFastRank优化。
TopN优化方法
无排名优化
TopN的输出结果无需要显示rownum值，仅需在最终前端显式时进行1次排序，极大地减少输入结果表的数据量。无排名优化方法详情请参见TopN语句。
增加TopN的Cache大小
TopN为了提升性能有一个State Cache层，Cache层能提升对State的访问效率。TopN的Cache命中率的计算公式为。
cache_hit = cache_size*parallelism/top_n/partition_key_num
例如，Top100配置缓存10000条，并发50，当您的PatitionBy的key维度较大时，例如10万级别时，Cache命中率只有10000*50/100/100000=5%，命中率会很低，导致大量的请求都会击中State（磁盘），性能会大幅下降。因此当PartitionKey维度特别大时，可以适当加大TopN的CacheS ize，相对应的也建议适当加大TopN节点的Heap Memory（请参见
手动配置调优
）。
##默认10000条，调整TopN cahce到20万，那么理论命中率能达200000*50/100/100000 = 100%。
blink.topn.cache.size=200000
PartitionBy的字段中要有时间类字段
例如每天的排名，要带上Day字段。否则TopN的结果到最后会由于State ttl有错乱。
3、高效去重方案
TopN算法
当TopN的输入是非更新流（例如Source），TopN只有一种算法AppendRank。当TopN的输入是更新流时（例如经过了AGG/JOIN计算），TopN有3种算法，性能从高到低分别是：UpdateFastRank 、 UnaryUpdateRank和RetractRank。算法名字会显示在拓扑图的节点名字上。
UpdateFastRank ：最优算法。
需要具备2个条件：
输入流有PK信息。
排序字段的更新是单调的，且单调方向与排序方向相反。例如，ORDER BY COUNT/COUNT_DISTINCT/SUM（正数）DESC（仅实时计算2.2.2及以上版本支持）。
如果您要获取到优化Plan，需要告诉系统total_fee
insert
into print_test
SELECT
cate_id,
seller_id,
stat_date,
pay_ord_amt --不输出rownum字段，能减小结果表的输出量。
FROM (
SELECT
*,
ROW_NUMBER () OVER (
PARTITION BY cate_id,
stat_date --注意要有时间字段，否则state过期会导致数据错乱。
ORDER
BY pay_ord_amt DESC
) as rownum --根据上游sum结果排序。
FROM (
SELECT
cate_id,
seller_id,
stat_date,
--重点。声明Sum的参数都是正数，所以Sum的结果是单调递增的，因此TopN能使用优化算法，只获取前100个数据。
sum (total_fee) filter (
where
total_fee >= 0
) as pay_ord_amt
FROM
random_test
WHERE
total_fee >= 0
GROUP
BY cate_name,
seller_id,
stat_date
) a
WHERE
rownum <= 100
);
UnaryUpdateRank：仅次于UpdateFastRank的算法。需要具备1个条件：输入流中存在PK信息。例如，ORDER BY AVG。
RetractRank：普通算法，性能最差，不建议在生产环境使用该算法。请检查输入流是否存在PK信息，如果存在，则可进行UnaryUpdateRank或UpdateFastRank优化。
TopN优化方法
无排名优化
TopN的输出结果无需要显示rownum值，仅需在最终前端显式时进行1次排序，极大地减少输入结果表的数据量。无排名优化方法详情请参见TopN语句。
增加TopN的Cache大小
TopN为了提升性能有一个State Cache层，Cache层能提升对State的访问效率。TopN的Cache命中率的计算公式为。
cache_hit = cache_size*parallelism/top_n/partition_key_num 例如，Top100配置缓存10000条，并发50，当您的PatitionBy的key维度较大时，例如10万级别时，Cache命中率只有10000*50/100/100000=5%，命中率会很低，导致大量的请求都会击中State（磁盘），性能会大幅下降。因此当PartitionKey维度特别大时，可以适当加大TopN的CacheS ize，相对应的也建议适当加大TopN节点的Heap Memory（请参见
手动配置调优
）。
##默认10000条，调整TopN cahce到20万，那么理论命中率能达200000*50/100/100000 = 100%。
blink.topn.cache.size=200000PartitionBy的字段中要有时间类字段
例如每天的排名，要带上Day字段。否则TopN的结果到最后会由于State ttl有错乱。
说明 仅实时计算3.2.1及以上版本支持高效去重方案。
实时计算的源数据在部分场景中存在重复数据，去重成为了用户经常反馈的需求。实时计算有保留第一条（Deduplicate Keep FirstRow）和保留最后一条（Deduplicate Keep LastRow）2种去重方案。
语法
由于SQL上没有直接支持去重的语法，还要灵活的保留第一条或保留最后一条。因此我们使用了SQL的ROW_NUMBER OVER WINDOW功能来实现去重语法。去重本质上是一种特殊的TopN。
SELECT *
FROM (
SELECT *,
ROW_NUMBER() OVER ([PARTITION BY col1[, col2..]
ORDER BY timeAttributeCol [asc|desc]) AS rownum
FROM table_name)
WHERE rownum = 1参数 说明 ROW_NUMBER() 计算行号的OVER窗口函数。行号从1开始计算。 PARTITION BY col1[, col2..] 可选。指定分区的列，即去重的KEYS。 ORDER BY timeAttributeCol [asc|desc]) 指定排序的列，必须是一个时间属性的字段（即Proctime或Rowtime）。可以指定顺序（Keep FirstRow）或者倒序 （Keep LastRow）。 rownum 仅支持 rownum=1或rownum<=1。 如上语法所示，去重需要两层Query：
使用
ROW_NUMBER()窗口函数来对数据根据时间属性列进行排序并标上排名。
说明
- 当排序字段是Proctime列时，Flink就会按照系统时间去重，其每次运行的结果是不确定的。
- 当排序字段是Rowtime列时，Flink就会按照业务时间去重，其每次运行的结果是确定的。
对排名进行过滤，只取第一条，达到了去重的目的。
说明 排序方向可以是按照时间列的顺序，也可以是倒序：
Deduplicate Keep FirstRow
保留首行的去重策略：保留KEY下第一条出现的数据，之后出现该KEY下的数据会被丢弃掉。因为STATE中只存储了KEY数据，所以性能较优，示例如下。
SELECT *
FROM (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) as rowNum
FROM T
)
WHERE rowNum = 1说明 以上示例是将T表按照b字段进行去重，并按照系统时间保留第一条数据。Proctime在这里是源表T中的一个具有Processing Time属性的字段。如果您按照系统时间去重，也可以将Proctime字段简化
PROCTIME()函数调用，可以省略Proctime字段的声明。Deduplicate Keep LastRow
保留末行的去重策略：保留KEY下最后一条出现的数据。保留末行的去重策略性能略优于LAST_VALUE函数，示例如下。
SELECT *
FROM (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) as rowNum
FROM T
)
WHERE rowNum = 1说明 以上示例是将T表按照b和d字段进行去重，并按照业务时间保留最后一条数据。Rowtime在这里是源表T中的一个具有Event Time属性的字段。
4、高效的内置函数使用内置函数替换自定义函数
实时计算的内置函数在持续的优化当中，请尽量使用内部函数替换自定义函数。实时计算2.0版本对内置函数主要进行了如下优化：
- 优化数据序列化和反序列化的耗时。
- 新增直接对字节单位进行操作的功能。
KEY VALUE函数使用单字符的分隔符
KEY VALUE 的签名：KEYVALUE(content, keyValueSplit, keySplit, keyName)，当keyValueSplit和KeySplit是单字符（例如，冒号（:）、逗号（,））时，系统会使用优化算法，在二进制数据上直接寻找所需的keyName 的值，而不会将整个content做切分。性能约提升30%。
多KEY VALUE场景使用MULTI_KEYVALUE
说明 仅实时计算2.2.2及以上版本支持MULTI_KEYVALUE。
在Query中对同一个Content进行大量KEY VALUE的操作，会对性能产生很大影响。例如Content中包含10个Key-Value对，如果您希望把10个Value的值都取出来作为字段，您就需要写10个KEY VALUE函数，则系统就会对Content进行10次解析，导致性能降低。
在这种情况下，建议您使用MULTI_KEYVALUE表值函数，该函数可以对Content只进行一次Split解析，性能约能提升50%~100%。
LIKE操作注意事项
- 如果需要进行StartWith操作，使用
LIKE 'xxx%'。 - 如果需要进行EndWith操作，使用
LIKE '%xxx'。 - 如果需要进行Contains操作，使用
LIKE '%xxx%'。 - 如果需要进行Equals操作，使用
LIKE 'xxx'，等价于str = 'xxx'。 - 如果需要匹配
_字符，请注意要完成转义LIKE '%seller/id%' ESCAPE '/'。_在SQL中属于单字符通配符，能匹配任何字符。如果声明为LIKE '%seller_id%'，则不单会匹配seller_id还会匹配seller#id、sellerxid或seller1id 等，导致结果错误。
慎用正则函数（REGEXP）
正则表达式是非常耗时的操作，对比加减乘除通常有百倍的性能开销，而且正则表达式在某些极端情况下可能会进入无限循环
- REGEXP
- REGEXP_EXTRACT
- REGEXP_REPLACE
5、网络传输的优化
使用内置函数替换自定义函数
实时计算的内置函数在持续的优化当中，请尽量使用内部函数替换自定义函数。实时计算2.0版本对内置函数主要进行了如下优化：
- 优化数据序列化和反序列化的耗时。
- 新增直接对字节单位进行操作的功能。
KEY VALUE函数使用单字符的分隔符
KEY VALUE 的签名：KEYVALUE(content, keyValueSplit, keySplit, keyName)，当keyValueSplit和KeySplit是单字符（例如，冒号（:）、逗号（,））时，系统会使用优化算法，在二进制数据上直接寻找所需的keyName 的值，而不会将整个content做切分。性能约提升30%。
多KEY VALUE场景使用MULTI_KEYVALUE
说明 仅实时计算2.2.2及以上版本支持MULTI_KEYVALUE。
在Query中对同一个Content进行大量KEY VALUE的操作，会对性能产生很大影响。例如Content中包含10个Key-Value对，如果您希望把10个Value的值都取出来作为字段，您就需要写10个KEY VALUE函数，则系统就会对Content进行10次解析，导致性能降低。
在这种情况下，建议您使用MULTI_KEYVALUE表值函数，该函数可以对Content只进行一次Split解析，性能约能提升50%~100%。
LIKE操作注意事项
- 如果需要进行StartWith操作，使用
LIKE 'xxx%'。 - 如果需要进行EndWith操作，使用
LIKE '%xxx'。 - 如果需要进行Contains操作，使用
LIKE '%xxx%'。 - 如果需要进行Equals操作，使用
LIKE 'xxx'，等价于str = 'xxx'。 - 如果需要匹配
_字符，请注意要完成转义LIKE '%seller/id%' ESCAPE '/'。_在SQL中属于单字符通配符，能匹配任何字符。如果声明为LIKE '%seller_id%'，则不单会匹配seller_id还会匹配seller#id、sellerxid或seller1id等，导致结果错误。
慎用正则函数（REGEXP）
正则表达式是非常耗时的操作，对比加减乘除通常有百倍的性能开销，而且正则表达式在某些极端情况下可能会进入无限循环
- REGEXP
- REGEXP_EXTRACT
- REGEXP_REPLACE
目前常见的Partitioner策略包括：
KeyGroup/Hash：根据指定的Key分配。
Rebalance：轮询分配给各个Channel。
Dynamic-Rebalance：根据下游负载情况动态选择分配给负载较低的Channel。
Forward：未Chain一起时，同Rebalance。Chain一起时是一对一分配。
Rescale：上游与下游一对多或多对一。
使用Dynamic-Rebalance替代Rebalance
Dynamic-Rebalance可以根据当前各Subpartition中堆积的Buffer的数量，选择负载较轻的Subpartition进行写入，从而实现动态的负载均衡。相比于静态的Rebalance策略，在下游各任务计算能力不均衡时，可以使各任务相对负载更加均衡，从而提高整个作业的性能。例如，在使用Rebalance时，发现下游各个并发负载不均衡时，可以考虑使用Dynamic-Rebalance。参数：
task.dynamic.rebalance.enabled=true， 默认关闭。使用Rescale替代Rebalance
说明 仅实时计算2.2.2及以上版本支持Rescale。
例如，上游是5个并发，下游是10个并发。当使用Rebalance时，上游每个并发会轮询发给下游10个并发。当使用Rescale时，上游每个并发只需轮询发给下游2个并发。因为Channel个数变少了，Subpartition的Buffer填充速度能变快，能提高网络效率。当上游的数据比较均匀时，且上下游的并发数成比例时，可以使用Rescale替换Rebalance。参数：
enable.rescale.shuffling=true，默认关闭。
6、推荐的优化配置方案
综上所述，作业建议使用如下的推荐配置。
# EXACTLY_ONCE语义。
blink.checkpoint.mode=EXACTLY_ONCE
# checkpoint间隔时间，单位毫秒。
blink.checkpoint.interval.ms=180000
blink.checkpoint.timeout.ms=600000
# 2.x使用niagara作为statebackend，以及设定state数据生命周期，单位毫秒。
state.backend.type=niagara
state.backend.niagara.ttl.ms=129600000
# 2.x开启5秒的microbatch。
blink.microBatch.allowLatencyMs=5000
# 整个Job允许的延迟。
blink.miniBatch.allowLatencyMs=5000
# 单个batch的size。
blink.miniBatch.size=20000
# local 优化，2.x默认已经开启，1.6.4需手动开启。
blink.localAgg.enabled=true
# 2.x开启PartialFina优化，解决COUNT DISTINCT热点。
blink.partialAgg.enabled=true
# union all优化。
blink.forbid.unionall.as.breakpoint.in.subsection.optimization=true
# object reuse优化，默认已开启。
#blink.object.reuse=true
# GC优化（SLS做源表不能设置该参数）。
blink.job.option=-yD heartbeat.timeout=180000 -yD env.java.opts='-verbose:gc -XX:NewRatio=3 -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:ParallelGCThreads=4'
# 时区设置。
blink.job.timeZone=Asia/Shanghai
二、Flink性能调优1、自动配置调优
实时计算 Flink新增自动调优功能autoconf。能够在流作业以及上下游性能达到稳定的前提下，根据您作业的历史运行状况，重新分配各算子资源和并发数，达到优化作业的目的。
1、首次智能调优
创建一个作业。
上线作业。选择智能推荐配置，指定使用CU数为系统默认，不填即可。点击下一步。
数据检查，预估消耗CU数。
在运维界面启动作业，根据实际业务需要指定读取数据时间。
说明:实时计算作业启动时候需要您指定启动时间。实际上就是从源头数据存储的指定时间点开始读取数据。指定读取数据时间需要在作业启动之前。例如，设置启动时间为1小时之前。
待作业稳定运行10分钟后，且以下状态符合要求，即可开始下一次性能调优。
- 运行信息拓扑图中IN_Q不为100%。
- 数据输入RPS符合预期。
- 运行信息拓扑图中IN_Q不为100%。
2、非首次性能调优
- 停止>下线作业。
- 重新上线作业。选择智能推荐配置，指定使用CU数为系统默认，不填即可。点击下一步。
- 数据检查，再次预估消耗CU数。
- 在运维界面启动作业，待作业稳定运行十分钟后，即可再一次性能调优。
说明：
- 自动配置调优一般需要3到5次才能达到理想的调优效果。请完成首次性能调优后，重复非首次性能调优过程多次。
- 每次调优前，请确保足够的作业运行时长，建议10分钟以上。
- 指定CU数（参考值） = 实际消耗CU数*目标RPS/当前RPS。
- 实际消耗CU数：上一次作业运行时实际消耗CU
- 目标RPS：输入流数据的实际RPS（或QPS）
- 当前RPS：上一次作业运行时实际的输入RPS
2、手动配置调优
手动配置调优可以分以下三个类型。
- 资源调优
- 作业参数调优
- 上下游参数调优
1、资源调优
资源调优即是对作业中的Operator的并发数（parallelism）、CPU（core）、堆内存（heap_memory）等参数进行调优。
1、分析定位资源调优节点
定位性能瓶颈节点
性能瓶颈节点为Vertex拓扑图最下游中参数IN_Q值为100%的一个或者多个节点。如下图，7号节点为性能瓶颈节点。 
分析性能瓶颈因素
性能瓶颈的可分为三类。
- 并发（parallelism）不足
- CPU（core）不足
- MEM（heap_memory）不足
如下图，7号节点的性能瓶颈是资源（CPU和/或MEM）配置不足所导致。 
说明：判断性能瓶颈因素方法
- 瓶颈节点的资源健康分为100，则认为资源已经合理分配，性能瓶颈是并发数不足所导致。
- 瓶颈节点的资源健康分低于100，则认为性能瓶颈是单个并发的资源（CPU和/或MEM）配置不足所导致。
- 无持续反压，但资源健康分低于100，仅表明单个并发的资源使用率较高，但暂不影响作业性能，可暂不做调优。
通过作业运维页面中Metrics Graph功能，进一步判断性能瓶颈是CPU不足还是MEM不足。步骤如下。
- 运维界面中，点击TaskExecutor，找到性能瓶颈节点ID，点击查看详情。
- 选择Metrics Graph，根据曲线图判断CPU或者MEM是否配置不足（很多情况下两者同时不足）。
2、调整资源配置
完成了性能瓶颈因素判断后，点击开发>基本属性>跳转到新窗口配置，开始调整资源配置。 
批量修改Operator
点击GROUP框，进入批量修改Operator数据窗口。
说明：
- GROUP内所有的operator具有相同的并发数。
- GROUP的core为所有operator的最大值。
- GROUP的_memory为所有operator之和。
- 建议单个Job维度的CPU:MEM=1:4，即1个核对应4G内存。
配置修改完成后点击应用当前配置并关闭窗口。
单个修改Operator
- 点击Operator框，进入修改Operator数据窗口。
- 配置修改完成后点击应用当前配置并关闭窗口。
参数调整说明
您只需调整parallelism、core和heap_memory三个参数，即能满足大部分的资源调优需求。
- Parallelism
- source节点 资源根据上游Partition数来。例如source的个数是16，那么source的并发可以配置为16、8、4等。不能超过16。
- 中间节点 根据预估的QPS计算。对于数据量较小的任务，设置和source相同的并发度。QPS高的任务，可以配置更大的并发数，例如64、128、或者256。
- sink节点 并发度和下游存储的Partition数相关，一般是下游Partition个数的2~3倍。如果配置太大会导致数据写入超时或失败。例如，下游sink的个数是16，那么sink的并发最大可以配置48。
- Core 即CPU，根据实际CPU使用比例配置，建议配置值为0.25，可大于1。
- Heap_memory 堆内存。根据实际内存使用状况进行配置。
- 其他参数
- state_size：默认为0，group by、join、over、window等operator需设置为1。
- direct_memory：JVM堆外内存，默认值为0, 建议不要修改。
- native_memory：JVM堆外内存，默认值为0，建议修改为10MB。
- chainingStrategy：chain策略，根据实际需要修改。
2、作业参数调优
- 在开发页面的右侧选择作业参数。
- 输入调优语句。
| 优化 | 解决问题 | 调优语句 |
|---|---|---|
| MiniBatch | 提升吞吐，降低对下游压力仅对Group by有效。 | blink.miniBatch.allowLatencyMs=5000 blink.miniBatch.size=1000 |
| LocalGlobal | 优化数据倾斜问题 | blink.localAgg.enable=true |
| TTL | 设置State状态时间 | 1.x：state.backend.rocksdb.ttl.ms=129600000 2.x：state.backend.niagara.ttl.ms=129600000 其中，1.x 表示需显式开启，2.x 表示默认开启。 |
注意：添加或删除MiniBatch或LocalGlobal参数，job状态会丢失，修改值大小状态不会丢失。
3、上下游参数调优
实时计算 Flink可以在with参数内设置相应的参数，达到调优上下游存储性能的目的。
调优步骤：
- 进入作业的开发界面。
- 确定需要调优的上下游引用表的语句。
- 在with参数中配置相应的调优参数。如下图。
1、batchsize参数调优
实时计算 Flink的每条数据均会触发上下游存储的读写，会对上下游存储形成性能压力。可以通过设置batchsize，批量的读写上下游存储数据来降低对上下游存储的压力。
| 名字 | 参数 | 详情 | 设置参数值 |
|---|---|---|---|
| Datahub源表 | batchReadSize | 单次读取条数 | 可选，默认为10 |
| Datahub结果表 | batchSize | 单次写入条数 | 可选，默认为300 |
| 日志服务源表 | batchGetSize | 单次读取logGroup条数 | 可选，默认为10 |
| ADB结果表 | batchSize | 每次写入的批次大小 | 可选，默认为1000 |
| RDS结果表 | batchSize | 每次写入的批次大小 | 可选，默认为50 |
注意： 添加、修改或者删除以上参数后，作业必须停止-启动后，调优才能生效。
2、cache参数调优
| 名字 | 参数 | 详情 | 设置参数值 |
|---|---|---|---|
| RDS维表 | Cache | 缓存策略 | 默认值为None,可选LRU、ALL。 |
| RDS维表 | cacheSize | 缓存大小 | 默认值为None,可选LRU、ALL。 |
| RDS维表 | cacheTTLMs | 缓存超时时间 | 默认值为None,可选LRU、ALL。 |
| OTS维表 | Cache | 缓存策略 | 默认值为None, 可选LRU,不支持ALL。 |
| OTS维表 | cacheSize | 缓存大小 | 默认值为None, 可选LRU,不支持ALL。 |
| OTS维表 | cacheTTLMs | 缓存超时时间 | 默认值为None, 可选LRU,不支持ALL。 |
注意： 添加、修改或者删除以上参数后，作业必须停止-启动后，调优才能生效。
4、手动配置调优流程
- 资源调优、作业参数调优、上下游参数调优
- 开发上线作业
- 资源配置方式：使用上次资源配置
- 数据检查
- 上线
说明：完成资源、作业参数、上下游参数调优后，手动配置调优后续的步骤与自动配置调优基本一致。区别在于资源配置环节需要选择使用上次资源配置。
3、FAQ1、Q：性能调优后作业为什么运行不起来？
A:可能性1：首次自动配置时指定了CU数，但指定的CU数太小（比如小于自动配置默认算法的建议值，多见于作业比较复杂的情况），建议首次自动配置时不指定CU数。 可能性2：默认算法建议的CU数或指定的CU数超过了项目当前可用的CU数，建议扩容。
2、Q：Vertex拓扑中看不到持续反压，但延迟却非常大，为什么？
A：可能性1：若延时直线上升，需考虑是否上游source中部分partition中没有新的数据，因为目前delay统计的是所有partition的延时最大值。 可能性2：Vertex拓扑中看不到持续反压，那么性能瓶颈点可能出现在source节点。因为source节点没有输入缓存队列，即使有性能问题，IN_Q也永远为0（同样，sink节点的OUT_Q也永远为0）。 解决方案：通过手动配置调优，将source节点（GROUP）中的operator中chainingStrategy修改为HEAD，将其从GROUP中拆解出来。然后上线运行后再看具体是哪个节点是性能瓶颈节点，若依然看不到性能瓶颈节点，则可能是因为上游source吞吐不够，需考虑增加source的batchsize或增加上游source的shard数。 
3、Q:如何判断持续反压，反压时如何判断性能瓶颈点？
A:Vertex拓扑中某些节点的IN_Q持续为100%则存在持续反压情况，最后一个(或多个)IN_Q为100%的节点为性能瓶颈点。如下示例： 
上图存在反压,瓶颈在6号节点。
上图存在反压，瓶颈在2号节点。
上图存在反压，瓶颈在8号节点。
上图可能存在节点，瓶颈在0号节点。
4、Q: 如何判断数据倾斜？
A：(1)表象上看，某些节点不论增加多大的并发仍存在性能瓶颈，则可能存在数据倾斜。 (2)在Vertex拓扑中点击疑似存在数据倾斜的节点（一般为性能瓶颈节点），进入SubTask List界面，重点观察RecvCnt和InQueue，若各ID对应的RecvCnt值差异较大（一般超过1个数量级）则认为存在数据倾斜，若个别ID的InQueue长期为100%，则认为可能存在数据倾斜。 解决方案：请您参看GROUP BY 数据出现热点、数据倾斜。 
5、Q: 上线时指定15CU，但是上线后实际消耗仅为10CU，什么原因？
A：这种问题一般发生在Vertex只有一个节点的情况，此时由于source上游的物理表的shard数为1，Flink要求source的并发不能超过上游shard数，导致无法增加并发，因此亦无法增加到指定的CU数。 解决方案：
- 增加上游物理表的shard数。
- 将ID0的节点中的operator拆开，将source节点（GROUP）中的operator chainingStrategy修改为HEAD，将其从GROUP中拆解出来，然后手动配置调优。
6、Q: 上线时出现如左上图的告警，或出现诸如“Cannot set chaining strategy on Union Transformation”错误，如何处理？
A：这是由于作业的SQL有改动，导致DAG改变。 解决方案：通过重新获取配置解决，开发-基本属性-跳转到新窗口配置-重新获取配置信息。 
