pyflink的table api 能否根据 filed的值进行判断插入到不同的sink表中 ？
使用pyflink的table api 能否根据 filed的值进行判断插入到不同的sink表中？
场景：使用pyflink通过filter进行条件过滤后插入到sink中， 比如以下两条消息，logType不同，可以使用filter接口进行过滤后插入到sink表中： { "logType":"syslog", "message":"sla;flkdsjf" } { "logType":"alarm", "message":"sla;flkdsjf" } t_env.from_path("source")
.filter("logType=syslog")
.insert_into("sink1") 有方法直接在上面的代码中通过判断logType字段的类型实现类似if else的逻辑吗： if logType=="syslog": insert_into(sink1) elif logType=="alarm": insert_into(sink2)
如果insert_into 有.filter .select等接口的返回值的话就好办了，可以接着往下通过filter进行判断，代码类似以下：
t_env.from_path("source")
.filter("logType=syslog")
.insert_into("sink1")
.filter("logType=alarm")
.insert_into("sink2") 请各位大牛指点，感谢
*来自志愿者整理的flink邮件归档
写回答
-
Table API 不用 if/else 直接用类似逻辑即可：
val t1 = table.filter('x > 2).groupBy(..) val t2 = table.filter('x <= 2).groupBy(..) t1.insert_into("sink1) t2.insert_into("sink2")
*来自志愿者整理的flink邮件归档
2021-12-06 21:20:57赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
