测试使用如下结构:
table= t_env.from_path("source")
if table.filter("logType=syslog"):
table.filter("logType=syslog").insert_into("sink1")
elif table.filter("logType=alarm"):
table.filter("logType=alarm").insert_into("sink2")
我测试了下,好像table.filter("logType=syslog").insert_into("sink1")生效,下面的elif不生效,原因是table.filter("logType=syslog")或者table.where在做条件判断的同时已经将数据进行过滤,走不到下面的分支??
在 2020-06-19 10:08:25,"jack" <wslyk...@163.com> 写道:
>使用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")
>请各位大牛指点,感谢
>
>
>
>
>