RocketMQ Connect 構建數據流轉處理平臺_傅玉寶.pdf

1、傅玉寶Apache RocketMQ Committer阿里云消息混合云負責人RocketMQ Connect RocketMQ Connect 構建數據流轉處理平臺構建數據流轉處理平臺Contents01RocketMQ Connect 概述02RocketMQ Connect 實現原理03RocketMQ Connect 與OceanBase構建數據流轉平臺04RocketMQ Connect 現狀與未來目 錄05總結01RocketMQ Connect 概述削峰填谷異步解耦通過業務系統的松耦合設計，即便下游系統出現不可用甚至宕機，都不會影響到核心系統的正常運轉，同時由于上下游異步解耦，可

2、以靈活高效的適應因業務快速發展而帶來的變化，如新系統擴容。秒殺、搶紅包等大型活動時會正在短時間內帶來較高的流量，如果沒做相應的保護而導致系統超負荷甚至崩潰，或因限制太過導致請求大量失敗而影響用戶體驗，削峰填谷是解決該問題的有效方式。RocketMQ Connect 概述Apache RocketMQ 介紹事務消息交易、紅包等場景需要確保數據的最終一致性，大量引入 RocketMQ 的分布式事務，既可以實現系統之間的解耦，又可以保證數據的一致性。RocketMQ Connect 概述RocketMQ 豐富的數據生態SourcesDestinationRocketMQ Connect 概述沒有 C

3、onnect 的數據同步方案OB2ESData syncProducerServiceConsumerService方案一方案二Data PipelineRocketMQ Connect 概述沒有 Connect，數據同步面臨的挑戰數據源多 上百數據源，n*n 問題 系統差異性較大 研發周期長高可用Failover 能力當一個節點掛掉是否這個節點的任務就停止了，任務重新啟動是否還可以斷點續傳集群運維集群監控，運維管理，隨著數據管道的增多，如何管理，運維監控這些數據管道也變的越來越復雜高性能如何高效的從源數據源到目的數據源的數據同步彈性擴縮容根據系統流量動態的增加減少節點數量，既能通過擴容滿足高

4、峰期業務，也能在低峰期縮減節點，節省成本RocketMQ Connect 概述構建 RocketMQ Connect 數據集成解決方案1.標準化數據集成 API（Openmessaging Connect API）（標準數據同步格式，數據及 Schema，任務模型抽象）。2.標準化運行時-Connect Runtime。3.基于 Connector API 生態擴展能力。CDC流計算數據湖MySQLOceanBaseRedisESHivePostgreSQLDirosMongoDBHudiRocketMQ Connect Runtime集群管理配置管理位點管理負責均衡REST APIhttp:

5、/127.0.0.1:8081/connectors/OceanbaseConnector1?config=connector.class:org.apache.rocketmq.connect.mysql.OceanbaseSourceConnector,connect.topicname:table1Topic,jdbc:/Users/connect/sink-file1.txt,username:root,password:123456,table:table1http:/127.0.0.1:8081/connectors/ESSinkConnector1?config=connecto

6、r.class:org.apache.rocketmq.connect.es.ESSinkConnector,connecttopicnames:table1Topic,jdbc:/Users/connect/sink-file1.txt,username:root,password:123456,table:table1配置化使用RocketMQ Connect 概述Connect 方案的優勢 模型統一，基于 Connect 框架，標準化接入各種數據源。插件式，易添加 Connctor，成本低。高可用，自動容錯處理。易擴展，動態擴縮容。易運維，提供統一集群管理、監控的能力。RocketMQ

7、Connect 概述Connect 典型應用場景 新舊系統數據遷移（例如：MySQL-OceanBase）構建 Streaming Pipeline 數據通道 構建數據湖，搜索引擎，ETL 平臺等等 基于 Replicator 實現 RocketMQ 多活容災02RocketMQ Connect 實現原理RocketMQ Connect 原理High Level 架構Connect RuntimeSourceConnectorSink Connectortask2task1Connect Runtime:配置管理、負載均衡、任務調度。Worker：運行 Connector 實例和Task的線程

8、，同時包括對應的運行環境。Rest API 服務Connector：定義數據的來源和目標，并確定任務的數量，從Worker接收配置傳遞給同步任務。pulltask2task1pull數據流控制流RocketMQ Connect 原理集群部署 Connect 容錯機制Source Connector1Source Connector2w1t2w1t1w2t2w2t1Source Connector3w3t1w3t1worker1worker2worker3RocketMQ Connect 原理集群部署 Connect 容錯機制Source Connector1Source Connector2w

9、1t2w1t1w2t2w2t1Source Connector3w3t1w3t1worker1worker2worker3宕機RocketMQ Connect 原理集群部署 Connect 容錯機制Source Connector1Source Connector2w1t2w1t1w2t2w2t1Source Connector3w3t1w3t1worker1worker2worker3w3t1w3t2RocketMQ Connect 原理Rest API POST/connectors/connector name GET/connectors/connector name/config G

10、ET/connectors/connector name/status POST/connectors/connector name/stop03RocketMQ Connect 與OceanBase構建數據流轉平臺RocketMQ Connect 與 StreamingStreaming 典型應用場景業務1業務2業務3ETL數據應用 事件驅動型應用，由事件觸發計算、狀態更新或進行外部操作。如金融反詐、實時推薦、在線監控。數據分析型應用，對原始數據進行提取有價值的信息，進行分析。如企業BI、疫情防控。業務系統數據存儲數據流RocketMQ Connect 與 OceanBase 構建數據流轉平

11、臺基于 Connect 構建 RocketMQ Streaming業務1業務2業務3Connect業務系統數據存儲數據流RocketMQStreaming業務系統04RocketMQ Connect 現狀與未來Preview 階段OpenMessaging Connect API 1.0RocketMQ Connect Runtime 實現10+Connecotor 實現（MySQL，Redis，Kafka，Jms，MongoDB）RocketMQ Connect 現狀與未來RocketMQ 數據集成發展歷程1.0 階段OpenMessaging Connect API 2.0升級；Rocke

12、tMQ Connect Runtime 升級（Schema，Transform 特性支持）30+Connecotor 實現（CDC,JDBC，SFTP，NoSQL，緩存，HTTP,AMQP，JMS,數據湖，數倉，Replicator，Kafka ConnectorAdaptor）2.0 階段100+Connector 生態RocketMQ Connect 現狀與未來RocketMQ 數據集成發展歷程總結一個分布式、可擴展，容錯，高可靠的數據集成組件，實現 RocketMQ 與其它數據系統之間數據的流入流出，通過 RocketMQ Connect 可以實現 CDC，構建數據湖，結合流計算實現數據價值。RocketMQ ConnectThank you!RocketMQ 用戶調研GitHub:/oceanbase/服務號：OceanBase 數據庫星球論壇：