博客
关于我
Flink资料合集
阅读量:666 次
发布时间:2019-03-15

本文共 2690 字,大约阅读时间需要 8 分钟。

官方文档:https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/sql/queries.html

需求一:每1分钟汇总一次当天汇总数据

需求产生的原因:在flinksql中,每来一条记录,就会触发一次写数据库,记录来的太频繁,导致写入的数据库的记录数过多

解决方法:可以通过开启minibatch来实现

调整批量计算大小一般写在每个项目: 代号=init 的SQL里面```sql-- N秒计算一次,或是N条计算一次,调大可以提升处理效率set `table.exec.mini-batch.enabled`=`true`; set `table.exec.mini-batch.allow-latency`=`10s`;set `table.exec.mini-batch.size`=`100000`;
需求二:需要将每次汇总的数据持久化到存储中

需求产生的原因:在flinksql下,通过group by出来的数据都是CDC类型的(有delete、insert),但需要记录每次计算的结果,所以不能delete掉上一次计算group by出来的数据

set `table.dynamic-table-options.enabled`= `true`;  --支持动态定义表选项select .... from table_name/*+OPTIONS('maxwell-json-ext.ignore-delete'='true' --忽略delete操作,'maxwell-json-ext.update-to-insert'='true' --将update转成insert数据 )*/--忽略delete操作
问题一:mysql中的tinyint类型映射到flink上报错

在这里插入图片描述

解决方式:
编辑表的数据库连接,添加参数 tinyInt1isBit=false
例如:url=jdbc:mysql://br_live_prop_3323_mydb.lzdb.com:3323/live_prop?useCursorFetch=true&defaultFetchSize=5000&tinyInt1isBit=false
参考文档:https://www.jianshu.com/p/6885cad1cb14

需求三:自动生成测试数据:flink 1.11 中的随机数据生成器- DataGen connector

详见资料

https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/datagen.html

需求四:分割字符串,将分割后的结果达成多列:自定义表函数

第一步、自定义表函数继承TableFunction

import org.apache.flink.table.annotation.DataTypeHint;import org.apache.flink.table.annotation.FunctionHint;import org.apache.flink.table.api.*;import org.apache.flink.table.functions.TableFunction;import org.apache.flink.types.Row;import static org.apache.flink.table.api.Expressions.*;@FunctionHint(output = @DataTypeHint("ROW
"))public static class SplitFunction extends TableFunction
{ public void eval(String str) { for (String s : str.split(" ")) { // use collect(...) to emit a row collect(Row.of(s, s.length())); } }}

第二步、注册表函数,略

第三步、在flinksql中调用

SELECT myField, word, length FROM MyTable LEFT JOIN LATERAL TABLE(SplitFunction(myField)) ON TRUE

其他详情请参考官方文档:https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/functions/udfs.html

需求五:用NULL填充某个字段
hive里的用法 flink里的用法
null as user_id cast (null as bigint) as user_id
需求六:由于平台问题,项目无法正常启动,只能强行kill掉application,要怎么快速恢复数据
设置kafka的消费起始值:'scan.startup.mode'='timestamp','scan.startup.timestamp-millis'='1617551100000'   --从该kafka这一时间戳(毫秒值)记录的开始消费
问题七:Flink的背压问题

详情请见:https://blog.csdn.net/zc19921215/article/details/109246591

问题八:Flink SQL状态越来越多

详情请见:https://cloud.tencent.com/developer/article/1452854

问题九:Flink最终结果追加到mysql出现记录丢失的情况

问题排查方式:通过双写到hbase,对比hbase和mysql的记录,确定最终程序的计算结果没有问题。有问题的是写入mysql的过程中出现了问题,打印写mysql的日志,发现key值同设定的主键不一致,发现是在flink的平台上定义的mysql表主键定义有问题。

在这里插入图片描述
注意:在修改mysql主键的时候,除了要修改数据库里mysql的主键,同时要修改flink上定义的mysql表的主键,否则程序不会报错,但写入mysql会有数据丢失的问题!!!

你可能感兴趣的文章
mysql /*! 50100 ... */ 条件编译
查看>>
mudbox卸载/完美解决安装失败/如何彻底卸载清除干净mudbox各种残留注册表和文件的方法...
查看>>
mysql 1264_关于mysql 出现 1264 Out of range value for column 错误的解决办法
查看>>
mysql 1593_Linux高可用(HA)之MySQL主从复制中出现1593错误码的低级错误
查看>>
mysql 5.6 修改端口_mysql5.6.24怎么修改端口号
查看>>
MySQL 8.0 恢复孤立文件每表ibd文件
查看>>
MySQL 8.0开始Group by不再排序
查看>>
mysql ansi nulls_SET ANSI_NULLS ON SET QUOTED_IDENTIFIER ON 什么意思
查看>>
multi swiper bug solution
查看>>
MySQL Binlog 日志监听与 Spring 集成实战
查看>>
MySQL binlog三种模式
查看>>
multi-angle cosine and sines
查看>>
Mysql Can't connect to MySQL server
查看>>
mysql case when 乱码_Mysql CASE WHEN 用法
查看>>
Multicast1
查看>>
MySQL Cluster 7.0.36 发布
查看>>
Multimodal Unsupervised Image-to-Image Translation多通道无监督图像翻译
查看>>
MySQL Cluster与MGR集群实战
查看>>
multipart/form-data与application/octet-stream的区别、application/x-www-form-urlencoded
查看>>
mysql cmake 报错,MySQL云服务器应用及cmake报错解决办法
查看>>