StreamX

這種方法應(yīng)該是有問題的.行不通,原因如下:
1)在beginTransaction在恢復(fù)或者做快照之前調(diào)用,這里返回的是一個(gè)mysql的connection對象,在這里事務(wù)的載體就是這個(gè)connection對象
2)precommit是第一次預(yù)提交,如果你看過兩階段的源碼你就會(huì)發(fā)現(xiàn),這個(gè)是在給你傳入的那個(gè)事物的載體對象做快照,對應(yīng)你這里的就是mysql的connection,,接下來flink會(huì)將這個(gè)connection存儲(chǔ)到狀態(tài)后端,connection可不是一個(gè)普通對象,是一個(gè)數(shù)據(jù)庫的連接,能序列化成功?
3)commit,這個(gè)是快照成功了.flink內(nèi)部已經(jīng)將你上次的事務(wù)的載體對象給保存完畢了,在notifyCheckpointComplete里回調(diào)用戶的commit方法,通知用戶checkpoint成功了

connection根本就不能序列化,這是其一,
其二:即便這種方式可行,也會(huì)非常不穩(wěn)定,因?yàn)槊恳淮蝐ommit的提交都是一個(gè)checkpoin的過程,checkpoint的時(shí)間有長有短,如果遇到數(shù)據(jù)量大,checkpoint的時(shí)間長鹏倘，那就意味著你在beginTransaction創(chuàng)建的這個(gè)鏈接得一直是打開的狀態(tài),只到checkpoin完成才可以,那你得設(shè)置足夠長的connectionTimeOut時(shí)間和socketTimeOut時(shí)間
其三:.如果flink某次任務(wù)失敗了,你是從checkpoint中恢復(fù)上一個(gè)mysql的連接?

Flink實(shí)現(xiàn)Kafka到Mysql的Exactly-Once

Flink實(shí)現(xiàn)Kafka到Mysql的Exactly-Once 背景 最近項(xiàng)目中使用Flink消費(fèi)kafka消息，并將消費(fèi)的消息存儲(chǔ)到mysql中顽爹，看似一個(gè)很簡單的需求纤泵，在網(wǎng)...

it_zzy

10944 7 31