hi，大家好 对于json schema的问题，我想问一个其他的问题， 比如我要做一个实时报警系统，需要消费kafka的json数据来进行实时报警，我的想法是对于每一个报警都生成一个flink任务，主要报警逻辑翻译成一个flink sql。
其中kafka里面的json数据，每一个字段都是可以生成报警条件的，比如有一个json格式的header字段，这个字段里面的内容是不固定的， 某一个用户想用header.aaa字段，另一个用户想用header.bbb字段，比如每分钟header.aaa的count值大于100就报警。 这种情况下，我该如何定义我的schema呢？大家有没有什么想法，谢谢。 Benchao Li <libenc...@apache.org> 于2020年7月10日周五 下午1:54写道： > Hi Peihui, > > 正如Jark所说，FLINK-18002正是想解决这个问题，可以指定任意一个JsonNode为varchar类型。 > > 当然，这个feature不能解决所有问题，比如你有一个字段，内容不太确定，而且也不需要额外处理， > 主要是想保留这个字段，下游输出json的时候仍然还是这个字段。 > 如果用FLINK-18002的思路，输出到下游的时候，会把这部分数据整体作为一个json string，所以 > 从结果上来看，*还不能完全做到原封不动的输出到下游*。 > > 不知道后面这个场景是不是你面对的场景。如果是的话，我们目前有两个思路解决这个问题： > 1. 用RAW类型，这个需要json node类型对于flink来讲，都是可以序列化的 > 2. 用BINARY类型，因为现在已经有了对BINARY类型的处理，所以还需要额外加一个选项来指定对于BINARY类型 > 的处理模式。我们可以把任意json node转成它的json字符串表达形式，再转成byte[]进行中间的传输和处理；在 > 序列化的时候，再直接通过这个byte[]数据构造一个json node（这样可以保证它跟原来的json node一模一样）。 > > Jark Wu <imj...@gmail.com> 于2020年7月10日周五 下午12:22写道： > > > 社区有个 issue 正在解决这个问题，可以关注一下 > > https://issues.apache.org/jira/browse/FLINK-18002 > > > > Best, > > Jark > > > > On Fri, 10 Jul 2020 at 11:13, Leonard Xu <xbjt...@gmail.com> wrote: > > > > > Hi, Peihui > > > > > > 我理解你的需求是json中有一些复杂的字段，你不想解析，希望后续用UDTF在来解析，这个应该做不到的，现在的json format > > 的解析的底层实现 > > > 就是按照json的标准格式解析（jackson）的，没法将一个 > > > jsonObject解析成一个String。另外如果你jsonObject中的内容格式不确定，也不适合在Schema中声明， > > > 因为SQL 是预编译后执行的，不能做到schema里是三个field，执行时又能解析四个field。 > > > > > > 一种做法是定义复杂的jsonObject对应的ROW<a INT, b STRING, …> > > > 将全部可能的字段包含进去，每条记录没有的字段解析出来的会是null，fail-on-missing-field 默认关闭的， > > > 另外一种推荐你把复杂的字段在上游就转义成一个String放到json的一个field中，这样Flink解析出来就是一个String, > > > 然后query里用UDTF处理。 > > > > > > > > > 祝好 > > > Leonard Xu > > > > > > > > > > > > > > > > 在 2020年7月10日，10:16，Peihui He <peihu...@gmail.com> 写道： > > > > > > > > Hello， > > > > > > > > 实际上，我也并不太关心这个字段的内容，能按string 保存下来就好了。 > > > > > > > > Best wishes. > > > > > > > > Peihui He <peihu...@gmail.com> 于2020年7月10日周五 上午10:12写道： > > > > > > > >> Hello, > > > >> > > > >> 明白您的意思。但是当一个字段下的json 字段不确定，类似一个黑盒子一样的化，就不好定义了。 > > > >> > > > >> > > > >> Best wishes. > > > >> > > > >> LakeShen <shenleifight...@gmail.com> 于2020年7月10日周五 上午10:03写道： > > > >> > > > >>> Hi Peihui, > > > >>> > > > >>> 如果消费的 Kafka json 中，json 比较复杂的话，比如存在嵌套，就像下面的格式： > > > >>> > > > >>> { > > > >>> "a":"b", > > > >>> "c":{ > > > >>> "d":"e", > > > >>> "g":"f" > > > >>> } > > > >>> }， > > > >>> > > > >>> 那么在 kafka table source 可以使用 row 来定义: > > > >>> > > > >>> create table xxx ( > > > >>> a varchar, > > > >>> c row<d varchar , g varchar> > > > >>> ) > > > >>> > > > >>> 如果 还存在嵌套，可以继续再使用 Row 来定义。 > > > >>> > > > >>> Best, > > > >>> LakeShen > > > >>> > > > >>> Peihui He <peihu...@gmail.com> 于2020年7月10日周五 上午9:12写道： > > > >>> > > > >>>> Hello: > > > >>>> > > > >>>> 在用flink > > > sql从kafka消费数据时，有些json比较复杂，想直接定义为object，在后续通过udf转为string。 > > > >>>> > > > >>>> 有什么办法不？我在建表的时候有提示过json_object 类型，但是用了又提示其他类型。 > > > >>>> > > > >>>> > > > >>>> Best wishes. > > > >>>> > > > >>> > > > >> > > > > > > > > > > > -- > > Best, > Benchao Li >
