Flink问题之中字段访问报错如何解决-阿里云开发者社区

问题一：如何统计n小时内，flink成功从kafka消费的数据量？

Hi 各位,

我需要统计出flink最近 n小时(例如24小时？) 成功从kafka中消费的数据量，有什么比较好的方案吗？*来自志愿者整理的flink邮件归档

参考回答：

可以在metrics 上报时或落地前对source两次上报间隔的numRecordsOut值进行相减，最后呈现的时候按时间段累计就可以了

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359871

问题二：请教，有哪些连接器能让我获得一个非 update/delete，但又有主键的流呢？

背景：我想试用 flink sql 的 deduplicate 处理一个带主键的流，我发现

如果我使用 mysql-cdc 获得一个流，它会报错 Deduplicate doesn't support consuming update and delete changes....
如果我使用 kafka json 获得一个流，虽然 deduplicate 不报错，但是不能设置主键，报错 The Kafka table '...' with 'json' format doesn't support defining PRIMARY KEY constraint on the table
一个简单的满足上面要求的流是带主键的 datagen，它有主键，而且可以被 deduplicate 处理，但是太不可控。

请问还有别的连接器或格式支持吗，从文档中似乎不能很快地获得这一点。*来自志愿者整理的flink邮件归档

参考回答：

json格式改debezium-json试试

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/359872

问题三：用table api 提交多个SQL 的时候，有什么API 能提前检查SQL 的错误吗？

dear all : 用table api 提交多个SQL 的时候，有什么API 能提前检查SQL 的错误吗？每次都是提交执行的时候才报错。理论上里面在SQL解析部分就能发现，有现成的API吗？还是说需要自己去抓那部分解析代码，然后封装出API。如果没有，希望能提供这个功能，blink 应该是有的。*来自志愿者整理的flink邮件归档

参考回答：

可以看看 TableEnvironment#execute逻辑，如果能够将sql 转化成 Transformation，那么语法应该没有问题。

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/370022

问题四：flinksql 1.12.1 row中字段访问报错怎么办

hi, all 定义一个 ScalarFunction class Test extends ScalarFunction{ @DataTypeHint("ROW") def eval(): Row ={ Row.of("a", "b", "c") } }

当执行下面语句的时候 select Test().a from taba1 会报下面的错误：

anonfun$build$1.apply(NestedProjectionUtil.scala:112)atorg.apache.flink.table.planner.plan.utils.NestedProjectionUtilanonfun$build$1.apply(NestedProjectionUtil.scala:112)atorg.apache.flink.table.planner.plan.utils.NestedProjectionUtil

anonfun$optimize$1.apply(FlinkChainedProgram.scala:62)atorg.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgramanonfun$optimize$1.apply(FlinkChainedProgram.scala:62)atorg.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram

anonfun$optimize$1.apply(FlinkChainedProgram.scala:62) at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgramanonfunoptimizeoptimizeoptimize1.apply(FlinkChainedProgram.scala:58) at scala.collection.TraversableOnce

anonfun$foldLeft$1.apply(TraversableOnce.scala:157)atscala.collection.TraversableOnceanonfun$foldLeft$1.apply(TraversableOnce.scala:157)atscala.collection.TraversableOnce