Flink kafka自定义metrics在influxdb上解析失败怎么办？
大家好，Flink版本1.13.1。 我在FlinkKafkaProducer的invork方法中，创建了一个counter，它采用sinkTableName+_receive_count作为最终的metric 表名，并且采用influxDB作为上报终端，其中，sinkTableName是从catalog信息里拿到的，我从构造中一层层传递进去。
但是发现一个奇怪的问题：当我采用简单的形如insert into a select from b where b.c='d'的SQL语句时，influxDB中的表可以成功被建出来； 但如果加上UDF，比如 insert into a select CreateJsonOuter('tablename',CaptTimeSlewTime(Record.CAPTURE_TIME, 'YEAR_,MONTH'),'data',CreateJsonInner(true)) from source where Record.datasource='xx'时，influxDB就会报这样的解析错误，不知道大佬们有没有遇见过，困扰一天了！任何建议都不胜感激！ org.influxdb.InfluxDBException$UnableToParseException: partial write: unable to parse 'taskmanager_job_task_operator_xx_ODS_INFO_receive_count,host=xx,job_name= insert \ into\ ODS_INFO\ select \ CreateJsonOuter(' :missing fields unable to parse ''tablename', 'ODS_INFO',CaptTimeSlewTime(Record.CAPTURE_TIME, 'YEAR_,MONTH'),'data',CreateJsonInner(true,': missing fields)*来自志愿者整理的flink邮件归档
写回答
-
是通过什么方式拿到 sink table name 呢？从报错信息来看，拿到的可能并不是“a” 这样的 table name，而是一个 insert
语句的 digest，因此 metric 表名会变得比较复杂，导致解析错误。
当然也可以选择对 metric 表名进行转义。*来自志愿者整理的FLINK邮件归档
2021-12-02 15:10:09赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
