Hi,

Got it, seems this way is not flexable enough, but still thanks so much for 
your great support!  Good wished!




Regards && Thanks
Hunk








At 2022-04-02 16:34:29, "Qingsheng Ren" <renqs...@gmail.com> wrote:
>Hi,
>
>If the schema of records is not fixed I’m afraid you have to do it in UDTF. 
>
>Best, 
>
>Qingsheng
>
>> On Apr 2, 2022, at 15:45, wang <24248...@163.com> wrote:
>> 
>> Hi,
>> 
>> Thanks for your quick response! 
>> 
>> And I tried the format "raw", seems it only support single physical column, 
>> and in our project reqiurement, there are more than one hundred columns in 
>> sink table. So I need combine those columns into one string in a single UDF?
>> 
>> Thanks && Regards,
>> Hunk
>> 
>> 
>> 
>> 
>> 
>> 
>> 
>> At 2022-04-02 15:17:14, "Qingsheng Ren" <renqs...@gmail.com> wrote:
>> >Hi,
>> >
>> >You can construct the final json string in your UDTF, and write it to Kafka 
>> >sink table with only one field, which is the entire json string constructed 
>> >in UDTF, and use raw format [1] in the sink table:
>> >
>> >CREATE TABLE TableSink (
>> >    `final_json_string` STRING
>> >) WITH (
>> >    ‘connector’ = ‘kafka’,
>> >    ‘format’ = ‘raw’,
>> >    ...
>> >)
>> >
>> >So the entire flow would be like:
>> >
>> >1. Kafka source table reads data
>> >2. UDTF parses the ‘content’ field, and construct the final json (id, 
>> >content without backslash) string you need, maybe using Jackson [2] or 
>> >other json tools
>> >3. Insert the constructed json string as the only field in sink table
>> >
>> >The key problem is that the schema of field “content” is not fixed, so you 
>> >have to complete most logics in UDTF. 
>> >
>> >[1] 
>> >https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/formats/raw/
>> >[2] https://github.com/FasterXML/jackson
>> >
>> >Best regards, 
>> >
>> >Qingsheng
>> >
>> >
>> >> On Apr 2, 2022, at 14:47, wang <24248...@163.com> wrote:
>> >> 
>> >> Hi,
>> >> 
>> >> Thanks so much for your support! 
>> >> 
>> >> But sorry to say I'm still confused about it. No matter what the udf 
>> >> looks like, the first thing I need confirm is the type of 'content' in 
>> >> TableSink, what's the type of it should be, should I use type Row, like 
>> >> this?
>> >> 
>> >>      CREATE TABLE TableSink (
>> >>           `id` STRING NOT NULL,
>> >>           `content` ROW<name STRING, age BIGINT>
>> >>      )
>> >>      WITH (
>> >>          ...
>> >>     );
>> >> 
>> >> This type is only suitable for source input {"schema": "schema_infos", 
>> >> "payload": {"id": "10000", "content": "{\"name\":\"Jone\",\"age\":20}"}}
>> >> 
>> >> But the json key name and format of 'content' in source is variable, if 
>> >> the source input is 
>> >> {"schema": "schema_infos", "payload": {"id": "10000", "content": 
>> >> "{\"color\":\"Red\",\"BackgroundColor\":\"Yellow\",\"Height\":30"}}
>> >> 
>> >> I should define `content` in TableSink with type `content` ROW<color 
>> >> STRING, BackgroundColor STRING, Height BIGINT>, like this:
>> >> 
>> >>      CREATE TABLE TableSink (
>> >>           `id` STRING NOT NULL,
>> >>           `content` ROW<color STRING, BackgroundColor STRING, Height 
>> >> BIGINT>
>> >>      )
>> >>      WITH (
>> >>          ...
>> >>     );
>> >> 
>> >> And in input json also might contains json array, like: 
>> >> {"schema": "schema_infos", "payload": {"id": "10000", "content": 
>> >> "{\"color\":\"Red\",\"BackgroundColor\":\"Yellow\",\"Height\":30",\"detail\":[{\"value_type\":1,\"value_name\":\"AAA\"},{\"value_type\":2,\"value_name\":\"BBB\"}]}}
>> >> 
>> >> 
>> >> So is there some common type I can use which can handle all input json 
>> >> formats?  
>> >> 
>> >> Thanks so much!!
>> >> 
>> >> 
>> >> Thanks && Regards,
>> >> Hunk
>> >> 
>> >> 
>> >> 
>> >> 
>> >> 
>> >> 
>> >> At 2022-04-01 17:25:59, "Qingsheng Ren" <renqs...@gmail.com
>> >> > wrote:
>> >> >Hi, 
>> >> >
>> >> >I’m afraid you have to use a UDTF to parse the content and construct the 
>> >> >final json string manually. The key problem is that the field “content” 
>> >> >is actually a STRING, although it looks like a json object. Currently 
>> >> >the json format provided by Flink could not handle this kind of field 
>> >> >defined as STRING. Also considering the schema of this “content” field 
>> >> >is not fixed across records, Flink SQL can’t use one DDL to consume / 
>> >> >produce records with changing schema. 
>> >> >
>> >> >Cheers,
>> >> >
>> >> >Qingsheng
>> >> >
>> >> >> On Mar 31, 2022, at 21:42, wang <
>> >> 24248...@163.com
>> >> > wrote:
>> >> >> 
>> >> >> Hi dear engineer,
>> >> >> 
>> >> >> Thanks so much for your precious time reading my email!
>> >> >> Resently I'm working on the Flink sql (with version 1.13) in my 
>> >> >> project and encountered one problem about json format data, hope you 
>> >> >> can take a look, thanks! Below is the description of my issue.
>> >> >> 
>> >> >> I use kafka as source and sink, I define kafka source table like this:
>> >> >> 
>> >> >>      CREATE TABLE TableSource (
>> >> >>           schema STRING,
>> >> >>           payload ROW(
>> >> >>               `id` STRING,
>> >> >>               `content` STRING
>> >> >>          )
>> >> >>      )
>> >> >>      WITH (
>> >> >>          'connector' = 'kafka',
>> >> >>          'topic' = 'topic_source',
>> >> >>          'properties.bootstrap.servers' = 'localhost:9092',
>> >> >>          '
>> >> properties.group.id
>> >> ' = 'all_gp',
>> >> >>          'scan.startup.mode' = 'group-offsets',
>> >> >>          'format' = 'json',
>> >> >>          'json.fail-on-missing-field' = 'false',
>> >> >>          'json.ignore-parse-errors' = 'true'
>> >> >>      );
>> >> >> 
>> >> >> Define the kafka sink table like this:
>> >> >> 
>> >> >>      CREATE TABLE TableSink (
>> >> >>           `id` STRING NOT NULL,
>> >> >>           `content` STRING NOT NULL
>> >> >>      )
>> >> >>      WITH (
>> >> >>          'connector' = 'kafka',
>> >> >>          'topic' = 'topic_sink',
>> >> >>          'properties.bootstrap.servers' = 'localhost:9092',
>> >> >>          'format' = 'json',
>> >> >>          'json.fail-on-missing-field' = 'false',
>> >> >>          'json.ignore-parse-errors' = 'true'
>> >> >>     );
>> >> >> 
>> >> >> 
>> >> >> Then insert into TableSink with data from TableSource:
>> >> >>     INSERT INTO TableSink SELECT id, content FROM TableSource;
>> >> >> 
>> >> >> Then I use "kafka-console-producer.sh" to produce data below into 
>> >> >> topic "topic_source" (TableSource):
>> >> >>     {"schema": "schema_infos", "payload": {"id": "10000", "content": 
>> >> >> "{\"name\":\"Jone\",\"age\":20}"}}
>> >> >> 
>> >> >> 
>> >> >> Then I listen on topic_sink (TableSink) by kafka-console-consumer.sh, 
>> >> >> the output is:
>> >> >>     {"id":"10000","content":"{\"name\":\"Jone\",\"age\":20}"}
>> >> >> 
>> >> >> But what I want here is {"id":"10000","content": 
>> >> >> {"name":"Jone","age":20}}
>> >> >> I want the the value of "content" is json object, not json string.
>> >> >> 
>> >> >> And what's more, the format of "content" in TableSource is not fixed, 
>> >> >> it can be any json formated(or json array format) string, such as:
>> >> >> {"schema": "schema_infos", "payload": {"id": "10000", "content": 
>> >> >> "{\"color\":\"Red\",\"BackgroundColor\":\"Yellow\",\"Height\":30"}}
>> >> >> 
>> >> >> 
>> >> >> So my question is, how can I transform json format string(like 
>> >> >> "{\"name\":\"Jone\",\"age\":20}")  from TableSource to json object 
>> >> >> (like{"name":"Jone","age":20} ).
>> >> >> 
>> >> >> 
>> >> >> Thanks && Regards,
>> >> >> Hunk
>> >> >> 
>> >> >> 
>> >> >> 
>> >> >> 
>> >> >>  
>> >> 
>> 
>> 
>> 
>>  

Reply via email to