com.alibaba.ververica        

            flink-connector-mysql-cdc        

                      1.1          .          0                  

public                     static                     void                     main(String[] args)           throws                     Exception {        

              SourceFunction sourceFunction = MySQLSource.builder()        

                .hostname(          "localhost"          )        

                .port(          3306          )        

                .databaseList(          "test"          )        

                .tableList(          "test"          )        

                .deserializer(          new                     StringDebeziumDeserializationSchema())        

                .build();        

              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        

              env.addSource(sourceFunction)        

                .print().setParallelism(          1          );        

              env.execute();        

            }        

create                     table                     test(        

id           INT          ,        

name                     varchar          (100),        

description           varchar          (100),        

weight           DECIMAL          (10,3)        

)        

CREATE                     TABLE                     mysql_test (        

           id           INT                     NOT                     NULL          ,        

           name                     STRING,        

           description STRING,        

           weight           DECIMAL          (10,3)        

)           WITH                     (        

           'connector'                     =           'mysql-cdc'          ,        

           'hostname'                     =           'localhost'          ,        

           'port'                     =           '3306'          ,        

           'username'          =          'root'          ,        

           'password'          =          'root'          ,        

           'database-name'                     =           'test'          ,        

           'table-name'                     =           'test'        

);        

SELECT                     id,           UPPER          (          name          ), description, weight           FROM                     mysql_test;