5、金風-Flink流批一體在小米的實踐.pdf

1、金風/小米軟件開發工程師 Flink流批一體在小米的實踐小米的大數據發展演變流批一體的平臺層支持流批一體典型應用場景未來規劃#1#2#3#4#1小米的大數據發展演變發展演變201920202019前離線計算為Spark實時計算為SparkStreaming，少部分Storm開始接入使用Flink,廣泛運用于搜索推薦，廣告實時樣本,實時ETL等場景2021開始接入使用FlinkSQL，廣泛運用于實時數倉構建，實時ETL作業開始接入數據湖，基于數據湖打造流批一體的解決方案2019現狀計算資源存儲YarnHiveIcebergHBaseTalosKuduDoris實時離線FlinkSpark Str

2、eamingFlinkSparkHiveFlink JarFlink SQLSparkFlink SQL BatchSpark Jar Spark SQL Hive SQL未來實時離線FlinkSpark StreamingFlinkSparkHiveFlink JarFlink SQLSparkFlink SQL BatchSpark Jar Spark SQL Hive SQLFlink實時離線Flink SQLFlinkFlink SQL流批一體化#2流批一體的平臺建設流批一體的平臺建設基于metacat構建統一的元數據管理元數據基于ranger構建統一的權限管理權限統一離線作業調度和實

3、時調度調度豐富離線和實時的生態生態統一的元數據管理MetacatHive RDSKuduDorisTalos(小米自研消息隊列)FlinkSparkPrestoHive統一的元數據管理SERVICE_CLUSTER .DATABASE .TABLE大多數系統以服務名和集群名作為一級catalog和系統原本的database對應，如果沒有database(es,talos(消息隊列)，則默認使用default和系統原本的table對應，es則是對應索引名，消息隊列則對應topic名INSERT INTO iceberg_cluster2.db1.tb1SELECT f1,f2,f3 FROM t

4、alos_cluster1.default.topic1;一條簡單的SQL，即可編寫一個數據入湖的作業！統一的權限管理RangerHiveIcebergKuduDorisMySQLFlink SQLSQLSource/SinkFieldsRangerAdminPhysicalPlanFlink JarTable APIDataStreamSource/Sink TableINSERT INTO iceberg_cluster2.db1.tb1SELECT f1,f2,f3 FROM talos_cluster1.default.topic1;Source:talos_cluster1.defa

5、ult.topic/f1,f2,f3Sink:iceberg_cluster2.db1.tb1/f4,f5,f6 統一的元數據+權限一站式開發體驗靈活的權限管理統一離線和實時調度離線調度實時調度批流混合INSERT INTO iceberg_cluster2.db1.tb1SELECT f1,f2,f3 FROM iceberg_cluster1.default.topic1;batch作業，默認全量讀取iceberg表，全量讀取完之后，作業結束.Streaming作業，默認全量讀取iceberg表，并實時消費增量的數據.batch+streaming,先啟動一個batch作業，默認全量讀取i

6、ceberg表，讀完之后batch作業結束.然后啟動一個streaming作業，實時消費增量的數據.生態建設Flink 社區HiveIcebergHBaseTalosKuduDorisFlink Bahir其他社區小米內部RedisRocketMQJDBCPegasusElasticSearchFalconOpenTSDBDruidHybridSourceCDC-SINKMySQLCDC生態建設Iceberg批量讀寫流式入湖增量消費以batch形式讀寫iceberg支持CDC數據實時寫入iceberg增量消費iceberg表，包括v1/v2表社區已支持社區已支持小米內部實現生態建設Hybrid

7、 SourceCREATE TABLE hybridSource(f1 VARBINARY)WITH(connector=hybrid,table-list=mysql_xxxxx.dbname1.tabled+;talos_xxx.default.topic1 );對接內部catalog 按順序讀區配置的source表生態建設CDC SinkCREATE TABLE sink_table(value VARBINARY)WITH(connector=cdc-sink,table=iceberg_cluster1.iceberg.binlog_sink_table,-真實要寫入的表 table

8、-filter=mysql_10086_dbname1.dbname1.tabled+;需要過濾的表，分號分割);對接內部catalog 按table-filter進行表的過濾 統一處理DDL變更消息#3流批一體應用場景數據集成離線集成實時集成批流混合集成數據集成-離線集成HiveJdbcIcebergDorisTiDBFlinkSQLBatchData XHiveJdbcIcebergDorisTiDB更加豐富的Source&Sink 并發處理能力字段映射降低了多平臺維護壓力數據集成-實時集成Talos（小米自研消息隊列）IcebergTiDBHiveJdbcIcebergDorisTiDB

9、FlinkSQLLCSSinkTiCDC Flink MySQL-CDC MySQLBinlogLCS Binlog研Binlog收集服務，類似canal內部基于SparkStreaming 的數據集成系統數據集成-批流混合Talos（小米自研消息隊列）FlinkSQLMySQLHybrid SourceFlinkCDCMySQL按DB劃分，將不同的database數據寫到不同的TopicHiveJdbcIcebergDorisTiDB先批量讀MySQL 再實時消費Talos(消息隊列)數據集成-批流混合Talos（小米自研消息隊列）FlinkSQLTICDCTIDBFlinkSQLBatch

10、全量增量批作業實時作業先起批作業做全量導，再起流作業做增量導HiveJdbcIcebergDorisTiDB數倉架構升級傳統離線數倉Lambda實時數倉基于數據湖的實時數倉 保存全量數據 產出結果準確率高 T+1延遲 一般凌晨調度,對集群造成較大壓力 實時鏈路提供有效性支持 離線鏈路提供正確性支持 支持數據回溯，OLAP查詢 平臺維護兩套架構，運維成本高 業務維護兩套代碼，開發成本高 兩套鏈路產出結果可能不一致 簡化架構，降低了開發維護壓力.兼顧延遲和正確性，同時對OLAP有較好的支持 無法做到秒級延遲數倉架構升級小米實時數倉實踐 小米手機激活實時數倉小米銷售服務實時數倉小米手機激活實時數倉業

11、務流程激活數據采集提前激活數據清洗自然激活活躍小米手機激活實時數倉采集日志RDSTalos實時數倉離線數倉ODSDWDDWMDMDIMTalos+KuduHiveODSDWDDWMDMDIMHiveOLAP引擎數據應用HBase+FileSystemADS數倉架構小米銷售服務實時數倉銷售服務訂單物流商品售后門店小米銷售服務實時數倉采集日志RDSTalosODSDWDDWMDMDIMTalos+KuduTalos+FileSystem數倉架構-實時架構Talos離線Spark離線補充維度數據，防止狀態過期數倉架構升級小米數據湖數倉實踐 小米APP近實時數倉小米X業務數據鏈路重構小米APP近實時數

12、倉Talos數倉架構采集日志IcebergV1IcebergV2IcebergV2DIMT-1T-2ODSDWDDMFlink SQLFlink SQLFlink SQL Batch小米X業務數據鏈路重構KuduHiveIcebergV2IcebergV2Spark MergeInto產生實時流IcebergV2Flink SQL實時處理Flink SQL實時處理DIM舊鏈路改造，產生變更數據，并寫入數據湖58 分鐘延遲1分鐘延遲1分鐘延遲小米X業務數據鏈路重構KuduHiveIcebergV2IcebergV2Merge Into修正實時流IcebergV2Merge Into修正實時流Merge Into修正實時流MySQLTalosCDCTalosTalosChangelog-jsonChangelog-jsonFlinkSQL實時處理FlinkSQL實時處理實時架構：兼顧實時性+準確性#4未來規劃未來規劃 Flink SQL batch應用于更加復雜的場景 Flink Dynamic Table 跟進，結合消息隊列+數據湖，實現真正流批一體 Hybrid Source 升級，基于新版Source接口進行改造.