1. 健康飲食推薦系統(tǒng)的技術(shù)架構(gòu)設(shè)計(jì)這個(gè)基于Hadoop的健康飲食推薦系統(tǒng)采用了典型的Lambda架構(gòu)兼顧了批處理和實(shí)時(shí)處理的需求。系統(tǒng)整體分為三個(gè)主要層次數(shù)據(jù)采集層、數(shù)據(jù)處理層和應(yīng)用服務(wù)層。數(shù)據(jù)采集層負(fù)責(zé)從多個(gè)渠道獲取原始數(shù)據(jù)包括用戶基本信息年齡、性別、BMI等飲食記錄每日攝入食物種類和數(shù)量運(yùn)動(dòng)數(shù)據(jù)通過(guò)可穿戴設(shè)備采集健康指標(biāo)如血糖、血壓等醫(yī)療數(shù)據(jù)數(shù)據(jù)處理層是整個(gè)系統(tǒng)的核心我們采用了Hadoop生態(tài)系統(tǒng)中的多個(gè)組件HDFS作為分布式文件存儲(chǔ)系統(tǒng)YARN進(jìn)行資源管理MapReduce用于批量數(shù)據(jù)處理Spark提供實(shí)時(shí)計(jì)算能力HBase作為NoSQL數(shù)據(jù)庫(kù)存儲(chǔ)用戶畫像應(yīng)用服務(wù)層基于SpringBoot構(gòu)建提供了RESTful API接口供前端調(diào)用。系統(tǒng)架構(gòu)設(shè)計(jì)中特別考慮了以下幾個(gè)關(guān)鍵點(diǎn)數(shù)據(jù)分區(qū)策略按照用戶ID進(jìn)行哈希分區(qū)確保同一用戶的數(shù)據(jù)落在同一節(jié)點(diǎn)上減少跨節(jié)點(diǎn)查詢計(jì)算資源分配為Spark Streaming分配獨(dú)立的計(jì)算資源避免與批處理作業(yè)爭(zhēng)搶資源緩存機(jī)制使用Redis緩存熱門推薦結(jié)果減輕數(shù)據(jù)庫(kù)壓力提示在實(shí)際部署時(shí)建議為Hadoop和Spark配置不同的工作隊(duì)列通過(guò)YARN的Capacity Scheduler實(shí)現(xiàn)資源隔離這是很多初學(xué)開發(fā)者容易忽略的關(guān)鍵配置。2. 核心算法設(shè)計(jì)與實(shí)現(xiàn)2.1 用戶畫像構(gòu)建用戶畫像是推薦系統(tǒng)的基礎(chǔ)我們采用多維度特征提取方法// 用戶畫像特征提取示例代碼 public class UserProfile { private String userId; private MapString, Double nutrientPreferences; // 營(yíng)養(yǎng)素偏好 private MapString, Integer foodCategoryPref; // 食物類別偏好 private HealthCondition healthCondition; // 健康狀況 private ListAllergy allergies; // 過(guò)敏信息 // 特征計(jì)算方法 public void calculateFeatures(ListDietRecord records) { // 實(shí)現(xiàn)特征計(jì)算邏輯 } }2.2 協(xié)同過(guò)濾算法優(yōu)化傳統(tǒng)的協(xié)同過(guò)濾算法在飲食推薦場(chǎng)景面臨冷啟動(dòng)問題。我們的解決方案是混合推薦策略對(duì)于新用戶采用基于內(nèi)容的推薦根據(jù)用戶填寫的健康問卷對(duì)于有歷史數(shù)據(jù)的用戶采用改進(jìn)的協(xié)同過(guò)濾算法相似度計(jì)算優(yōu)化 使用改進(jìn)的余弦相似度計(jì)算用戶相似度加入了健康因素的權(quán)重相似度 α×飲食相似度 β×健康指標(biāo)相似度 γ×人口統(tǒng)計(jì)相似度實(shí)時(shí)反饋機(jī)制 通過(guò)Spark Streaming處理用戶的實(shí)時(shí)反饋如對(duì)推薦菜品的評(píng)分動(dòng)態(tài)調(diào)整推薦結(jié)果2.3 營(yíng)養(yǎng)均衡模型這是本系統(tǒng)的創(chuàng)新點(diǎn)之一我們?cè)O(shè)計(jì)了一個(gè)多目標(biāo)優(yōu)化模型目標(biāo)函數(shù) 最小化 Σ(實(shí)際攝入量 - 推薦攝入量)2 約束條件 1. 總熱量不超過(guò)每日需求 2. 各營(yíng)養(yǎng)素比例符合健康標(biāo)準(zhǔn) 3. 考慮用戶口味偏好 4. 排除過(guò)敏食材使用MapReduce實(shí)現(xiàn)該模型的并行計(jì)算// Map階段 public void map(LongWritable key, Text value, Context context) { // 計(jì)算單個(gè)食材的營(yíng)養(yǎng)貢獻(xiàn) } // Reduce階段 public void reduce(Text key, IterableFoodScore values, Context context) { // 匯總評(píng)分生成最終推薦列表 }3. 關(guān)鍵技術(shù)實(shí)現(xiàn)細(xì)節(jié)3.1 Hadoop集群配置優(yōu)化針對(duì)飲食推薦場(chǎng)景的特殊需求我們對(duì)Hadoop集群做了如下優(yōu)化HDFS配置塊大小設(shè)置為256MB默認(rèn)128MB副本數(shù)設(shè)置為3生產(chǎn)環(huán)境推薦值啟用Erasure Coding節(jié)省存儲(chǔ)空間YARN配置!-- yarn-site.xml 關(guān)鍵配置 -- property nameyarn.nodemanager.resource.memory-mb/name value16384/value !-- 根據(jù)實(shí)際機(jī)器配置調(diào)整 -- /property property nameyarn.scheduler.maximum-allocation-mb/name value12288/value /propertyMapReduce調(diào)優(yōu)設(shè)置合理的map和reduce任務(wù)數(shù)啟用Combiner減少網(wǎng)絡(luò)傳輸使用壓縮中間結(jié)果3.2 Spark實(shí)時(shí)處理實(shí)現(xiàn)實(shí)時(shí)推薦模塊使用Spark Streaming實(shí)現(xiàn)關(guān)鍵實(shí)現(xiàn)點(diǎn)包括數(shù)據(jù)接收val kafkaParams Map[String, Object]( bootstrap.servers - kafka1:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - diet_recommendation, auto.offset.reset - latest, enable.auto.commit - (false: java.lang.Boolean) ) val stream KafkaUtils.createDirectStream[String, String]( streamingContext, PreferConsistent, Subscribe[String, String](topics, kafkaParams) )狀態(tài)管理 使用mapWithState API維護(hù)用戶最近偏好def updateUserPreference(key: String, value: Option[UserAction], state: State[UserPreference]): Option[UserPreference] { // 實(shí)現(xiàn)狀態(tài)更新邏輯 }性能優(yōu)化調(diào)整批處理間隔通常2-5秒合理設(shè)置并行度使用Kryo序列化3.3 SpringBoot與Hadoop集成后端服務(wù)使用SpringBoot實(shí)現(xiàn)與Hadoop生態(tài)集成的關(guān)鍵點(diǎn)HBase連接配置Configuration public class HBaseConfig { Value(${hbase.zookeeper.quorum}) private String zookeeperQuorum; Bean public Connection hbaseConnection() throws IOException { Configuration config HBaseConfiguration.create(); config.set(hbase.zookeeper.quorum, zookeeperQuorum); return ConnectionFactory.createConnection(config); } }MapReduce作業(yè)提交public void submitJob() throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, DietRecommendation); job.setJarByClass(DietRecommendationJob.class); // 設(shè)置Mapper和Reducer job.setMapperClass(DietMapper.class); job.setReducerClass(DietReducer.class); // 設(shè)置輸入輸出路徑 FileInputFormat.addInputPath(job, new Path(inputPath)); FileOutputFormat.setOutputPath(job, new Path(outputPath)); // 提交作業(yè) boolean success job.waitForCompletion(true); if (!success) { throw new RuntimeException(Job failed); } }性能考慮使用連接池管理HBase連接異步執(zhí)行耗時(shí)操作實(shí)現(xiàn)合理的重試機(jī)制4. 系統(tǒng)部署與性能優(yōu)化4.1 集群部署方案我們建議的硬件配置方案節(jié)點(diǎn)類型數(shù)量CPU內(nèi)存存儲(chǔ)網(wǎng)絡(luò)Master216核64G1T SSD10GWorker532核128G4T HDD10GEdge18核32G500G SSD10G軟件版本選擇Hadoop 3.3.4Spark 3.2.1HBase 2.4.11JDK 1.8u301SpringBoot 2.6.7MySQL 5.7.364.2 性能測(cè)試與調(diào)優(yōu)我們對(duì)系統(tǒng)進(jìn)行了全面的性能測(cè)試關(guān)鍵指標(biāo)推薦響應(yīng)時(shí)間批處理模式平均2.3秒百萬(wàn)級(jí)用戶實(shí)時(shí)模式平均120毫秒吞吐量測(cè)試單Worker節(jié)點(diǎn)可處理5000 TPS整個(gè)集群可支持10萬(wàn)用戶并發(fā)資源利用率優(yōu)化通過(guò)動(dòng)態(tài)資源分配DRA提高資源利用率設(shè)置合理的Spark內(nèi)存分?jǐn)?shù)spark.memory.fraction0.6調(diào)整HBase讀寫緩存大小4.3 常見問題解決方案在實(shí)際部署中遇到的典型問題及解決方案數(shù)據(jù)傾斜問題癥狀某些Reduce任務(wù)執(zhí)行時(shí)間遠(yuǎn)長(zhǎng)于其他任務(wù)解決方案使用Salting技術(shù)分散熱點(diǎn)調(diào)整Partitioner在Spark中使用repartition小文件問題癥狀HDFS中存在大量小文件影響NameNode性能解決方案使用HAR文件歸檔實(shí)現(xiàn)小文件合并策略調(diào)整Hive/Spark輸出配置內(nèi)存溢出問題癥狀Spark作業(yè)頻繁出現(xiàn)OOM解決方案調(diào)整executor內(nèi)存配置減少單個(gè)分區(qū)的數(shù)據(jù)量使用更高效的數(shù)據(jù)結(jié)構(gòu)注意在升級(jí)Hadoop或Spark版本時(shí)務(wù)必先在測(cè)試環(huán)境驗(yàn)證兼容性特別是HDFS的版本兼容性經(jīng)常會(huì)導(dǎo)致生產(chǎn)環(huán)境問題。我們?cè)騂adoop 3.3.3的一個(gè)已知bug導(dǎo)致集群不穩(wěn)定回退到3.3.1后才解決。