> > *1. Is there some easy way to use deserialized JSON in DataStream without case classes or POJOs?* > Could you explain what you expected? Do you mean you want to just register a DataType that is able to bridge the received bytes to the POJO objects. No, heh, I don't want to have any POJO objects. I don't want users to have to write hardcoded java classes of our canonical JSONSchemas. I want someone to be able to use JSON data in Flink that we know conforms to a JSONSchema with types that map cleanly to Java types (i.e. no random additionalProperties and $refs) without hardcoding any duplicate information about that data that can be retrieved via our other internal API. (We have an HTTP 'schema registry' for JSONSchemas.).
Row (and RowData) can do this; I just want to use them easily with JSON in a DataStream. > *> 2. How can I use a DeserializationSchema<Row> to get a DataStream<Row> or even DataStreamSource<Row> in a unit test from either a file or String[]/byte[] of serialized JSON?* > For DeserializationSchema<RowData>, you can refer to the Kafka connector[2]. I think it should be similar to the DeserializationSchema<Row>. JsonRowDeserializationSchema is marked as deprecated and recommends to use the Table API. I can do that, but I feel like it is overkill for just wanting to use DataStream<Row>. I was trying to get away with starting and ending with the Table API always, where I can easily use DataType and RowData, but if I do some map transformations on the DataStream<Row> to produce a new stream, I need an explicitly declared TypeInformation<Row> that matches the new stream schema when converting back into the Table API. If I need to have the output TypeInformation<Row> explicitly declared anyway, I might as well just start with TypeInformation<Row> in the first place, and stay in DataStream world the whole time. FWIW, I think I've been able to accomplish what I was trying to do in this patch <https://gerrit.wikimedia.org/r/c/wikimedia-event-utilities/+/806319>. Still needs some testing, but I've written my own JSONSchema -> TypeInformation<Row> converter, and have copy/pasted and modified Flink's deprecated JsonRowDeserializaitonSchema into our code. Thank you for your responses! -Andrew Otto Wikimedia Foundation On Fri, Jun 17, 2022 at 12:33 AM Shengkai Fang <fskm...@gmail.com> wrote: > Hi. > > > *1. Is there some easy way to use deserialized JSON in DataStream > without case classes or POJOs?* > > Could you explain what you expected? Do you mean you want to just register > a DataType that is able to bridge the received bytes to the POJO objects. I > am not sure wether the current RAW type[1] in Flink Table is enough for > you. > > *> 2. How can I use a DeserializationSchema<Row> to get a DataStream<Row> > or even DataStreamSource<Row> in a unit test from either a file or > String[]/byte[] of serialized JSON?* > > For DeserializationSchema<RowData>, you can refer to the Kafka > connector[2]. I think it should be similar to the > DeserializationSchema<Row>. > > Best, > Shengkai > > [1] > https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/types/ > [2] > https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java#L234 > > > > Andrew Otto <o...@wikimedia.org> 于2022年6月17日周五 02:26写道: > >> At the Wikimedia Foundation, we use JSON and JSONSchemas for our events >> in Kafka. There are hundreds of these schemas and topics in Kafka. I'd >> like to provide library level integration between our 'Event Platform' JSON >> data and Flink. My main goal: >> >> *No case classes or POJOs. *The JSONSchemas should be enough. >> >> I can actually do this pretty easily with the Table API. I can >> convert from JSONSchema to a DataType, and then create a table with that >> DataType and format('json'). >> >> I'd like to be able to do the same for the DataStream API. From what I >> can tell, to do this I should be using a Row >> <https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/types/Row.html> >> as the record type. I can also convert from JSONSchema to >> TypeInformation<Row> pretty easily, using the Types factory >> <https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/api/common/typeinfo/Types.html> >> . >> >> While I can convert to and from >> <https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/data_stream_api/> >> the Table API to DataStream<Row>, it seems directly using DataStream<Row> >> of our JSON could be pretty useful, and would make it possible to use Flink >> without instantiating a StreamTableEnvironment or requiring a 'table >> planner'. Also, to convert back up to the Table API from a >> DataStream<Row>, I need the explicit TypeInformation<Row>, which I need to >> manually construct. >> >> Ah but, JsonRowDeserializationSchema >> <https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.html> >> is >> deprecated. Okay, fine I can copy it into my code and modify it for my >> purposes. But even if I do, I'm confused about something else: >> >> DeserializationSchema is not Table API specific (e.g. it can be used as >> the value deserializer in KafkaSource). Row is also not Table API specific >> (although I know the main purpose is to bridge Table to DataStream API). >> However, it seems that constructing a Source using DeserializationSchema is >> not really that common? KafkaSource uses it, but FileSource and >> env.fromElements don't? I'm trying to write integration tests for this >> that use the DataStream API. >> >> *tl;dr questions:* >> >> *1. Is there some easy way to use deserialized JSON in DataStream without >> case classes or POJOs?* >> >> *2. How can I use a DeserializationSchema<Row> to get a DataStream<Row> >> or even DataStreamSource<Row> in a unit test from either a file or >> String[]/byte[] of serialized JSON?* >> >> Thank you! >> >> >> >> >>