问题一：动态表 Change Log 格式
Flink 动态表的 Change Log 会有四种消息（INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE）. 其中 UPDATE_BEFORE 对应的是更新之前的完整记录，UPDATE_AFTER 对应的是更新之后的完整记录吗？ 我的问题特意强调完整记录，是为了确认 Change Log 的内容是更新后的完整数据，而不是 Set A = 'a' 这样的操作日志/更新语句。 此外，Delete语句对应的数据是完整记录还是操作日志呢？
这意味着Table Sink的时候，只需要获得INSERT， UPDATE_AFTER的数据，写入不支持UPSERT的存储。 并通过额外的逻辑判断来获得最新的数据是可行的。
-- *来自志愿者整理的flink邮件归档
参考答案：
是完整的记录。
upsert kafka 就是这样子实现的，只存储最新镜像。 但是有的 query 是会产生 delete 消息的，所以有时候还是需要存下 delete，像 upsert kafka 里就存成了kafka 的 tombstone 消息。*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370082?spm=a2c6h.13066369.question.46.33bf585f3ZrV23
问题二：SQL解析复杂JSON问题
麻烦问下我已经在字段上面定了结构，还需要再写format.json-schema吗？CREATE TABLE user_log( id VARCHAR, timestam VARCHAR, user_info ROW(user_id string, name string ), jsonArray ARRAY<ROW(user_id222 STRING, name222 STRING)> ) WITH ( 'connector.type' = 'kafka', 'connector.version' = 'universal', 'connector.topic' = 'complex_string', 'connector.properties.zookeeper.connect' = 'venn:2181', 'connector.properties.bootstrap.servers' = 'venn:9092', 'connector.startup-mode' = 'earliest-offset', 'format.type' = 'json', 'format.json-schema' = '{ "type": "object", "properties": { "id": {type: "string"}, "timestam": {type: "string"}, "user_info":{type: "object", "properties" : { "user_id" : {type:"string"}, "name":{type:"string"} } }, "jsonArray":{"type": "array", "items": { "type": "object", "properties" : { "user_id222" : {type:"string"}, "name222" : {type:"string"} } } } } }' );
*来自志愿者整理的flink邮件归档
参考答案：
Hi 你好,
这个取决于你使用的flink版本，1.11版本会自动从table schema中解析，而1.10版本如果table schema和json schema不是完全相同的话，需要手动写json-schema: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/json.html#data-type-mapping https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/json.html#data-type-mapping https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#json-format https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#json-format*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370083?spm=a2c6h.13066369.question.47.33bf585fQemfHB
问题三：一个关于实时合并数据的问题
想请教各位一个问题：目前有一个这样的需求：
数据流40W/s,数据有id，time，type....等字段，id有10亿个，现在想30分钟内，同一个id的信息只保存一条，时间的话要用事件的事件，不能用处理的时间 本人现在的思路是：根据id分组，然后做增量ck，状态信息存储每个id的最后的时间，然后每来一条数据会读取状态信息，然后做时间判断。但是发现这样做背压很高，数据消费很慢 请问各位，我这种思路是否可行？根据id分组会产生10亿个分组，这样会影响什么？还有其他更好的方法么？
谢谢各位解答疑惑！*来自志愿者整理的flink邮件归档
参考答案：
根据 id 去重并不意味着根据 id 做 keyby 分组，比如可以将 id 做个 mod 分成若干有限组，然后在 processFunction 中 MapState 存储 <id, 1> 进行去重处理
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370084?spm=a2c6h.13066369.question.48.33bf585fI2D86j
问题四：flinksql1.11长时间没有数据写入mysql，出问题
使用flinksql 1.11版本，运行计算好的指标落入mysql，长时间没有数据会报错，导致任务会失败。*来自志愿者整理的flink邮件归档
参考答案：
JdbcBatchingOutputFormat: for (int i = 0; i <= executionOptions.getMaxRetries(); i++) { try { attemptFlush(); batchCount = 0; break; } catch (SQLException e) { LOG.error("JDBC executeBatch error, retry times = {}", i, e); if (i >= executionOptions.getMaxRetries()) { throw new IOException(e); } try { if (!connection.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) { connection = connectionProvider.reestablishConnection(); jdbcStatementExecutor.closeStatements(); jdbcStatementExecutor.prepareStatements(connection); } } catch (Exception excpetion) { LOG.error("JDBC connection is not valid, and reestablish connection failed.", excpetion); throw new IOException("Reestablish JDBC connection failed", excpetion); } try { Thread.sleep(1000 * i); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); throw new IOException("unable to flush; interrupted while doing another attempt", e); } } }
嗯，看起来是这样的。
if (i >= executionOptions.getMaxRetries()) { throw new IOException(e); }这个判断重试的代码应该放在sleep 后面。不然，Caused by: java.io.IOException: java.sql.SQLException: No operations allowed after statement closed. 就没机会重建连接了。*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370085?spm=a2c6h.13066369.question.51.33bf585fijZlWy
问题五：请问flink sql实时计算分位数如何实现哈？
你好：
目前flink sql实时计算中没有percentile函数吗？如果没有，如何实现这一功能。 期待你的答复，谢谢！*来自志愿者整理的flink邮件归档
参考答案：
可以看下UDAF的文档： https://ci.apache.org/projects/flink/flink-docs-master/dev/table/functions/udfs.html#aggregate-functions *来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370087?spm=a2c6h.13066369.question.52.33bf585fnDag6n
