（4） Flink's table API and SQL table schema

List of articles

table schema Defined Flink Table The data table structure of , Include field names , type , meanwhile Table Schema Hui He Table Format Match , stay Table Data input or output process Schema Transformation

But when Table Input/Output Format and Table Schema When they disagree , All need corresponding Mapping Relationship to map

Be careful : Users must follow Input/Output The order of fields in the data source TableSchema

Log in and copy

       

        withSchema(
        
  new Schema()
        
  .field("id",Types.INT)
        
  .field("id",Types.STRING)
        
)
       
       

        
1.
        
2.
        
3.
        
4.
        
5.
       

In addition to creating Table Schema When specifying the name and type , Supported by proctime and rowtime And other methods to obtain the time attribute in the external data , among proctime Method does not need to pass parameters ,rowtime You need to define the time field and watermark Generative logic , At the same time, you can also use from Method maps from the dataset by name Table Schema Field information

Log in and copy

       

        .withSchema(
        
  new Schema()
        
  .field("Field",Types.INT)
        
  .proctime()
        
  .field("f2",Types.STRING)
        
  .rowtime(...)
        
  .field("f3",Types.STRING)
        
  .from("origin_field_name")
        
)
       
       

        
1.
        
2.
        
3.
        
4.
        
5.
        
6.
        
7.
        
8.
        
9.
       

If Table API be based on Event Time Time concept processing data , You need to generate time in the access data Rowtime Information , as well as Watermark The generative logic of

Log in and copy

       

        .rowtime(
        
  // You can extract from the input data according to the field name 
        
  new Rowtime().timestampsFromField("ts_filed")
        
  // Or from the bottom DataStream API From , The data source needs to support the allocation of timestamps ( Such as Kafka 0.10X)
        
  new Rowtime().timestampsFromSource()
        
  // Or through customization timestampsFromSource extract Rowtime
        
  new Rowtime().timestampsFromExtractor()
       
       

        
1.
        
2.
        
3.
        
4.
        
5.
        
6.
        
7.
       

Immediately after Rowtime() After the object instance, you need to specify Watermark Strategy

Log in and copy

       

        .rowtime(
        
// Time delay 2s Generate watermark
        
new Rowtime().watermarksPeriodicBounded(2000)
        
// and rowtime The maximum time is consistent 
        
new Rowtime().watermarksPeriodicAscending()
        
// Use the bottom layer DataStream API The built-in Watermark
        
new Rowtime().watermarksFromSource()
        
)
       
       

        
1.
        
2.
        
3.
        
4.
        
5.
        
6.
        
7.
        
8.