??????flink cdc ????mysql binlog?????? streaming????????????????????reload????????????????????

Tue, 27 Apr 2021 00:28:21 -0700 

????????????????????????


------------------ ???????? ------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<[email protected]&gt;;
????????:&nbsp;2021??4??27??(??????) ????3:15
??????:&nbsp;"[email protected]"<[email protected]&gt;;
????:&nbsp;"user-zh"<[email protected]&gt;;
????:&nbsp;??????flink cdc ????mysql binlog?????? 
streaming????????????????????reload????????????????????



????????????????datastreaming 
API??????debezium????????????????????????debezium??????
public static Properties debeziumProperties(){
&nbsp;&nbsp;&nbsp; Properties properties = new Properties();
properties.setProperty(??xxxx??,??xxxx");
&nbsp;&nbsp;&nbsp; return properties;
}


&nbsp;&nbsp;&nbsp; SourceFunction<T&gt; sourceFunction = 
MySQLSource.<T&gt;builder()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; . . .
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
.debeziumProperties(debeziumProperties())
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .build();


| |
MOBIN
|
|
[email protected]
|
??????????????????????


??2021??04??27?? 14:46????????????<[email protected]&gt; ??????
hi all

??????????flink cdc??????????????????streaming 
mode??????????binlog??????????????????????????????????????????????????mysql??RELOAD????????????????sql????????cdc????????????
debezium.snap.shot.locking.mode = none????????


????????streaming 
mode??????????????????????????????????????????????????reload??????????????flink 
cdc??????????????????????
Properties properties = new Properties();
properties.setProperty("debezium.snapshot.locking.mode", "none");
SourceFunction<String&amp;gt; sourceFunction = 
MySQLSource.<String&amp;gt;builder()
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .hostname("localhost")
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .port(port)
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .databaseList("database")&amp;nbsp;
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .tableList("database.test")
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .username(??user)
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .password("password")
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .debeziumProperties(properties)
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .deserializer(new 
StringDebeziumDeserializationSchema())
&amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; .build();

??.debeziumProperties(properties)????????????sql????????????????????????????????sourceFunction??????Streaming??????????????????????????????????????

回复