??????question

[????????]

I have found a way??
  select row(msg.f1) from table.

 

 ------------------ ???????? ------------------
  ??????: "????????"<2463...@qq.com>;
 ????????: 2019??9??4??(??????) ????10:57
 ??????: 
"????????"<2463...@qq.com>;"JingsongLee"<lzljs3620...@aliyun.com>;"user"<user@flink.apache.org>;
 
 ????: ??????question

 

 I want to output the query results to kafka, json format is as follows??
 {
 "id": "123",
 "serial": "6b0c2d26",
 "msg": {
  "f1": "5677"
 }
}
  How to define the format and schema of kafka sink??
 thanks!

 

 ------------------ ???????? ------------------
  ??????: "????????"<2463...@qq.com>;
 ????????: 2019??9??3??(??????) ????8:18
 ??????: "JingsongLee"<lzljs3620...@aliyun.com>;"user"<user@flink.apache.org>;
 
 ????: ??????question

 

thank you??  
  Let me try??

 

 ------------------ ???????? ------------------
  ??????: "JingsongLee"<lzljs3620...@aliyun.com>;
 ????????: 2019??9??3??(??????) ????7:53
 ??????: "????????"<2463...@qq.com>;"user"<user@flink.apache.org>;
 
 ????: Re:question

 

should be schema.field(??msg??, Types.ROW(...))?  And you should select msg.f1 
from table.
  

 Best
 Jingsong Lee
 

 


 ???????????? iPhone??
   ------------------Original Mail ------------------
 From:???????? <2463...@qq.com>
 Date:2019-09-03 09:22:41
 Recipient:user <user@flink.apache.org>
 Subject:question

  How do you do:
 My problem is flink table format and table schema mapping.
 The input data is similar to the following json format??
 { "id": "123", "serial": "6b0c2d26", "msg": { "f1": "5677" } } The format code 
for TableSource is as follows: new Json().schema(Types.ROW(new String[] { "id", 
"serial", "msg" }, new TypeInformation << ? > [] { Types.STRING(), 
Types.STRING(), Types.ROW(new String[] { "f1" }, new TypeInformation << ? > [] 
{ Types.STRING() }) }));
 

 The schema part of TableSource is as follows:
 Schema schema = new Schema(); schema.field("id", Types.STRING()); 
schema.field("serial", Types.STRING());
 

 I don't know how to define the f1 field of msg in the schema. I tried 
schema.field("f1", Types.STRING()) before; but I will report an error. What is 
the correct method?
 The following SQL can be run correctly:
 select id,serial,f1 from table; 
 

 My flink version is 1.8.1,use flink table & SQL API
 

 thanks;