簡(jiǎn)介這是一套基于Hadoop與Spark構(gòu)建的金融信貸風(fēng)險(xiǎn)控制系統(tǒng)源碼面向大數(shù)據(jù)開發(fā)、金融風(fēng)控算法及數(shù)據(jù)工程方向的開發(fā)者可用于理解海量借貸數(shù)據(jù)下的信用評(píng)估與欺詐檢測(cè)全流程。資源包共69個(gè)文件大小約72KB以36個(gè)Java文件、8個(gè)Scala文件組成核心計(jì)算邏輯配套12個(gè)XML配置及5個(gè)properties參數(shù)文件便于調(diào)整Spark/Hive運(yùn)行環(huán)境與模型參數(shù)另有SQL、JSON、JS等文件輔助數(shù)據(jù)集成與可視化展示。項(xiàng)目覆蓋數(shù)據(jù)接入、預(yù)處理、邏輯回歸/隨機(jī)森林等模型訓(xùn)練、風(fēng)險(xiǎn)評(píng)分及實(shí)時(shí)批處理結(jié)合等模塊源碼分層清晰適合以此為基礎(chǔ)擴(kuò)展實(shí)時(shí)風(fēng)控、特征工程或監(jiān)控告警等實(shí)踐。已有1219人學(xué)習(xí)下載對(duì)于希望掌握大數(shù)據(jù)風(fēng)控項(xiàng)目架構(gòu)與技術(shù)選型的開發(fā)者來說是一份可直接研讀和二次開發(fā)的參考實(shí)現(xiàn)。1. 基于Hadoop、Spark的大數(shù)據(jù)金融信貸風(fēng)險(xiǎn)控系統(tǒng)先看懂它的數(shù)據(jù)賬信貸風(fēng)控這個(gè)場(chǎng)景有個(gè)反直覺的結(jié)論真正決定系統(tǒng)能不能跑起來的不是模型有多先進(jìn)而是從申請(qǐng)信息、還款流水、外部征信到最終風(fēng)險(xiǎn)評(píng)分的這條數(shù)據(jù)鏈路是否通暢。我拿到一套大數(shù)據(jù)金融信貸風(fēng)險(xiǎn)控系統(tǒng)源碼第一件事從來不是翻模型代碼而是先看它的數(shù)據(jù)處理鏈路是怎么組織的?；贖adoop、Spark的大數(shù)據(jù)金融信貸風(fēng)險(xiǎn)控系統(tǒng)本質(zhì)是用HDFS做全量數(shù)據(jù)底座、用Hadoop生態(tài)做離線批處理、用Spark做高時(shí)效計(jì)算把信貸審批和貸后監(jiān)控變成一條可審計(jì)、可回溯的數(shù)據(jù)流水線。它能解決的是數(shù)據(jù)量大、特征維度多、審批時(shí)效要求高的信貸風(fēng)控場(chǎng)景。適合誰讀正在做風(fēng)控平臺(tái)技術(shù)選型、準(zhǔn)備把這類源碼落地成生產(chǎn)系統(tǒng)的工程師或者想系統(tǒng)理解數(shù)據(jù)鏈路與模型鏈路如何銜接的新手。2. 從源數(shù)據(jù)到風(fēng)險(xiǎn)分Hadoop 做數(shù)據(jù)底座、Spark 做計(jì)算引擎的架構(gòu)拆分2.1 分層設(shè)計(jì)離線層、近線層、在線層各管哪一段一套完整的信貸風(fēng)控系統(tǒng)數(shù)據(jù)鏈路一定不是一條直線而是按時(shí)效性拆成三個(gè)層次。離線層跑在Hadoop之上數(shù)據(jù)落在HDFS表結(jié)構(gòu)放在Hive里負(fù)責(zé)T1的全量批處理歷史申請(qǐng)數(shù)據(jù)歸檔、月度特征重算、模型訓(xùn)練樣本生成。近線層用Spark的微批或Structured Streaming負(fù)責(zé)分鐘級(jí)到小時(shí)級(jí)的增量特征刷新比如貸后行為監(jiān)控、額度復(fù)評(píng)需要用到的近幾小時(shí)交易匯總。在線層則是規(guī)則引擎和模型服務(wù)需要在毫秒級(jí)返回風(fēng)險(xiǎn)分這部分通常不直接跑Spark而是讀取預(yù)計(jì)算好的特征。這里有個(gè)常見誤解既然Spark既能做批處理又能做流處理為什么還要保留Hadoop答案是兩者根本不是替代關(guān)系。Hadoop提供的是分布式存儲(chǔ)HDFS和資源調(diào)度YARNSpark是運(yùn)行在YARN之上的計(jì)算框架。換句話說Hadoop是底座Spark是引擎。源碼里如果看到MRv2的作業(yè)或Hive UDF別急著說“這技術(shù)過時(shí)了”它們負(fù)責(zé)的是數(shù)據(jù)歸檔、全量重算這類對(duì)實(shí)時(shí)性不敏感但對(duì)穩(wěn)定性要求極高的任務(wù)。技術(shù)選型不是追新是讓每類任務(wù)落在它最合適的位置上。三層之間的流轉(zhuǎn)也很關(guān)鍵。離線層產(chǎn)出的特征寬表會(huì)被同步到在線層的緩存或特征庫(kù)中供實(shí)時(shí)查詢?cè)诰€層產(chǎn)生的審批結(jié)果和貸后表現(xiàn)又會(huì)回流到Hive表里成為下一輪模型訓(xùn)練的新樣本。這個(gè)回環(huán)結(jié)構(gòu)是信貸風(fēng)控系統(tǒng)和一般推薦系統(tǒng)最大的不同風(fēng)控系統(tǒng)的數(shù)據(jù)是自我進(jìn)化的每放一筆款、每產(chǎn)生一筆逾期都會(huì)成為未來模型的訓(xùn)練素材。2.2 特征加工鏈路Spark 做寬表 joinHadoop 做歸檔沉淀特征加工是風(fēng)控系統(tǒng)最重的一塊計(jì)算。申請(qǐng)人的基本信息、流水、多頭借貸記錄、歷史逾期記錄分散在幾十張表里必須拉到一起才能計(jì)算特征。常見做法是先用Spark SQL做多層寬表加工把明細(xì)數(shù)據(jù)聚合成以申請(qǐng)為粒度的特征表寬表里一行就是一個(gè)申請(qǐng)單對(duì)應(yīng)的幾百個(gè)特征這是后續(xù)模型訓(xùn)練和規(guī)則判斷的輸入。加工過程中Spark的優(yōu)勢(shì)體現(xiàn)得很明顯內(nèi)存計(jì)算避免了反復(fù)讀寫磁盤同一個(gè)口徑下Spark跑一個(gè)多表join的耗時(shí)通常只有MapReduce的幾分之一這對(duì)于需要按天重跑的全量特征任務(wù)來說直接決定了能不能在業(yè)務(wù)開門前跑完。Hadoop在這一環(huán)節(jié)的職責(zé)更多是數(shù)據(jù)沉淀。經(jīng)過Spark加工后的結(jié)果會(huì)寫回Hive表以分區(qū)形式組織按業(yè)務(wù)日期分區(qū)存放。這樣設(shè)計(jì)的價(jià)值在于一旦發(fā)現(xiàn)某個(gè)特征口徑算錯(cuò)了可以只重跑對(duì)應(yīng)分區(qū)而不需要?jiǎng)尤砹硗釮ive表配合HDFS的副本機(jī)制天然承擔(dān)了數(shù)據(jù)備份和災(zāi)難恢復(fù)的職責(zé)。生產(chǎn)系統(tǒng)里我一般會(huì)在離線層保留最近三年的原始明細(xì)數(shù)據(jù)壓縮后存放在HDFS上為后續(xù)的模型回溯分析留足彈藥。用Spark接增量明細(xì)、加工特征再落回Hive分區(qū)表的骨架大概長(zhǎng)這樣from pyspark.sql import SparkSession from pyspark.sql.functions import col, to_date, current_timestamp spark SparkSession.builder \ .appName(customer_feature_etl) \ .enableHiveSupport() \ .getOrCreate() biz_date 2025-01-15 df spark.read.format(parquet) \ .load(hdfs:///data/ods/customer_trans/dt{}.format(biz_date)) df_clean df.filter(col(trans_amt) 0) \ .withColumn(etl_time, current_timestamp()) \ .withColumn(feature_dt, to_date(col(trans_time))) df_clean.createOrReplaceTempView(trans_detail) spark.sql( SELECT customer_id, SUM(trans_amt) AS amt_total, COUNT(*) AS cnt_total, AVG(trans_amt) AS amt_avg FROM trans_detail GROUP BY customer_id ).write.mode(overwrite) \ .format(parquet) \ .partitionBy(feature_dt) \ .saveAsTable(dws.customer_feature_agg)代碼邏輯分四步先按業(yè)務(wù)日期讀取ODS層的客戶交易明細(xì)讀取時(shí)直接定位到當(dāng)天分區(qū)避免全表掃描然后做基礎(chǔ)清洗過濾掉金額為負(fù)的異常流水同時(shí)補(bǔ)上ETL時(shí)間和特征日期字段接著注冊(cè)成臨時(shí)視圖用Spark SQL做聚合最后按feature_dt分區(qū)寫入DWS層特征聚合表。這里有兩個(gè)參數(shù)值得留意一是read.load的路徑中dt{}這個(gè)日期參數(shù)應(yīng)當(dāng)由調(diào)度平臺(tái)傳入保證任務(wù)重跑時(shí)能指定正確分區(qū)二是write.mode(overwrite)對(duì)于按分區(qū)寫入的場(chǎng)景overwrite模式只覆蓋對(duì)應(yīng)分區(qū)不會(huì)影響其他日期的數(shù)據(jù)遠(yuǎn)比先刪表再重建安全。2.3 樣本與標(biāo)簽風(fēng)控建模里最容易翻車的時(shí)點(diǎn)對(duì)齊問題信貸風(fēng)控訓(xùn)練數(shù)據(jù)和普通機(jī)器學(xué)習(xí)訓(xùn)練數(shù)據(jù)最大的區(qū)別在于標(biāo)簽不是立刻能拿到的。某筆申請(qǐng)?jiān)?月15日通過審批它是不是壞樣本要看放款后90天甚至180天有沒有逾期這就是風(fēng)控里常說的表現(xiàn)期。因此源碼里樣本表的設(shè)計(jì)通常包含三個(gè)時(shí)間字段申請(qǐng)時(shí)間、放款時(shí)間、表現(xiàn)期截止時(shí)間。切樣本的時(shí)候只能取那些表現(xiàn)期已經(jīng)走完的歷史申請(qǐng)還在表現(xiàn)期內(nèi)的不能進(jìn)訓(xùn)練集。特征時(shí)點(diǎn)對(duì)齊是另一個(gè)重災(zāi)區(qū)。訓(xùn)練樣本里某個(gè)特征比如近三個(gè)月消費(fèi)金額計(jì)算窗口的截止點(diǎn)必須嚴(yán)格卡在申請(qǐng)時(shí)間點(diǎn)不能把申請(qǐng)之后的流水也算進(jìn)去。這個(gè)坑如果踩了模型在回測(cè)時(shí)的AUC會(huì)虛高上線后卻立刻失效。典型的錯(cuò)誤寫法是直接拿當(dāng)前時(shí)間作為窗口終點(diǎn)去算歷史樣本的特征這在源代碼評(píng)審時(shí)是最先要揪出來的問題。正確的做法是把申請(qǐng)時(shí)間作為參數(shù)傳入特征計(jì)算任務(wù)讓每一條樣本的特征都嚴(yán)格截取在它自己申請(qǐng)時(shí)間之前。還有一種常見做法是在數(shù)據(jù)表設(shè)計(jì)上做約束申請(qǐng)事實(shí)表和特征表都用申請(qǐng)單號(hào)做主鍵觀察期、表現(xiàn)期作為字段存下來。這樣不管是訓(xùn)練還是推理都從同一個(gè)寬表里取數(shù)避免訓(xùn)練和上線兩套口徑。寬表生成后模型訓(xùn)練只讀這張表不再回源明細(xì)既快又不容易出錯(cuò)。3. 讀懂這類源碼的四塊核心數(shù)據(jù)接入、特征計(jì)算、規(guī)則與模型、任務(wù)調(diào)度3.1 數(shù)據(jù)接入把申請(qǐng)信息與還款流水接進(jìn) Hive 的 Spark 作業(yè)骨架拿到源碼包先別急著跑先按目錄結(jié)構(gòu)把模塊劃分認(rèn)清楚。這類系統(tǒng)常見的模塊拆法是數(shù)據(jù)接入層、特征加工層、決策引擎層、管理后臺(tái)層。數(shù)據(jù)接入層負(fù)責(zé)把業(yè)務(wù)庫(kù)、消息隊(duì)列、外部數(shù)據(jù)源里的數(shù)據(jù)搬運(yùn)到HDFS和Hive里是整個(gè)鏈條的起點(diǎn)。信貸場(chǎng)景里的數(shù)據(jù)源有幾種典型形態(tài)業(yè)務(wù)申請(qǐng)數(shù)據(jù)存在業(yè)務(wù)庫(kù)中常用Sqoop或DataX同步還款流水和行為埋點(diǎn)走消息隊(duì)列外部征信數(shù)據(jù)以文件或接口方式提供。源碼里這一層通常是最穩(wěn)定的因?yàn)槟Ｊ较鄬?duì)固定但也是最先出問題的地方。以還款流水接入為例用Spark接消息隊(duì)列再落Hive分區(qū)的骨架代碼如下from pyspark.sql import SparkSession from pyspark.sql.functions import col, from_json, to_date from pyspark.sql.types import StructType, StringType, LongType, DoubleType spark SparkSession.builder \ .appName(repay_flow_ingest) \ .enableHiveSupport() \ .getOrCreate() schema StructType() \ .add(loan_id, StringType()) \ .add(repay_amt, DoubleType()) \ .add(repay_time, LongType()) \ .add(status, StringType()) raw spark.readStream.format(kafka) \ .option(kafka.bootstrap.servers, node01:9092,node02:9092) \ .option(subscribe, repay_flow) \ .option(startingOffsets, latest) \ .load() parsed raw.select( from_json(col(value).cast(string), schema).alias(data) ).select(data.*) parsed parsed.withColumn(dt, to_date(col(repay_time).cast(timestamp))) def write_batch(batch_df, batch_id): batch_df.write.mode(append) \ .format(parquet) \ .partitionBy(dt) \ .saveAsTable(dwd.repay_flow_daily) parsed.writeStream \ .foreachBatch(write_batch) \ .outputMode(append) \ .trigger(processingTime5 minutes) \ .start()這里用了Structured Streaming的foreachBatch模式每5分鐘把攢下的一批數(shù)據(jù)寫入Hive分區(qū)表。foreachBatch比直接writeStream到Hive更可靠因?yàn)榭梢栽谶@里加入去重、格式校驗(yàn)、甚至調(diào)用其他依賴該批數(shù)據(jù)的任務(wù)。參數(shù)上重點(diǎn)關(guān)注兩處startingOffsets設(shè)為latest表示只消費(fèi)新數(shù)據(jù)如果要做歷史補(bǔ)數(shù)需要改成earliest或指定offsettrigger的processingTime決定數(shù)據(jù)可見延遲5分鐘一刷對(duì)貸后監(jiān)控夠用但需要按業(yè)務(wù)對(duì)時(shí)效的要求動(dòng)態(tài)調(diào)整。3.2 特征計(jì)算窗口函數(shù)與時(shí)間滑窗的典型寫法特征加工是風(fēng)控源碼里代碼量最大的部分也是最值得逐行讀的部分。常見特征按窗口分為近1個(gè)月、近3個(gè)月、近6個(gè)月、近12個(gè)月甚至包含節(jié)假日在內(nèi)的特殊時(shí)段特征。這類時(shí)間滑窗特征在Spark SQL里非常適合用窗口函數(shù)加條件聚合來實(shí)現(xiàn)。近3個(gè)月還款行為特征的典型SQL寫法如下SELECT loan_id, customer_id, SUM(CASE WHEN repay_status overdue THEN 1 ELSE 0 END) AS overdue_cnt_3m, COUNT(*) AS repay_cnt_3m, SUM(repay_amt) AS repay_amt_3m, AVG(repay_amt) AS repay_avg_amt_3m, MAX(repay_amt) AS repay_max_amt_3m FROM dwd.repay_flow_daily WHERE repay_time TO_DATE(2025-01-15) - INTERVAL 3 MONTHS AND repay_time TO_DATE(2025-01-15) GROUP BY loan_id, customer_id這條SQL把還款明細(xì)按申請(qǐng)單聚合輸出逾期次數(shù)、還款筆數(shù)、總額、均值、最大值五個(gè)特征。需要注意的是WHERE條件里窗口終點(diǎn)的寫法用TO_DATE(2025-01-15)而不是CURRENT_DATE。如果用CURRENT_DATE跑歷史樣本的補(bǔ)數(shù)任務(wù)時(shí)就會(huì)把未來數(shù)據(jù)也算進(jìn)去導(dǎo)致訓(xùn)練集和線上特征口徑不一致。這個(gè)細(xì)節(jié)在多人協(xié)作的項(xiàng)目里反復(fù)出現(xiàn)也是絕大多數(shù)特征口徑事故的根源。另外特征計(jì)算任務(wù)最好做成配置驅(qū)動(dòng)窗口長(zhǎng)度、聚合字段、時(shí)間字段都放在配置表或配置項(xiàng)里代碼里不寫死。好處有兩個(gè)一是新增特征時(shí)不用改代碼改配置加一行就行二是模型迭代時(shí)方便對(duì)照特征清單審查口徑。源碼里如果特征都是硬編碼在一個(gè)大SQL里后續(xù)維護(hù)成本會(huì)很高拿到手后第一步建議就是把它重構(gòu)為配置化。3.3 規(guī)則引擎與模型打分規(guī)則表、PMML、原生 Pipeline 的取舍信貸風(fēng)控決策不是純模型而是規(guī)則與模型融合的結(jié)果。硬規(guī)則負(fù)責(zé)攔截監(jiān)管紅線和極端風(fēng)險(xiǎn)比如年齡超限、黑名單命中、身份證格式非法模型負(fù)責(zé)對(duì)剩余申請(qǐng)做精細(xì)化風(fēng)險(xiǎn)排序。源碼里面向這一塊的代碼通常分兩部分規(guī)則引擎和模型推理服務(wù)。規(guī)則引擎的設(shè)計(jì)常見做法是規(guī)則配置化規(guī)則不寫死在代碼里而是放在數(shù)據(jù)庫(kù)表或配置文件中支持熱更新。規(guī)則表的結(jié)構(gòu)一般包含這些字段CREATE TABLE risk_rule_set ( rule_id STRING, rule_name STRING, rule_expr STRING, -- 例如 score_model 600 AND apply_amt 50000 action STRING, -- PASS / MANUAL / REJECT priority INT, status STRING, update_time TIMESTAMP );rule_expr里存的是可解析的表達(dá)式規(guī)則引擎啟動(dòng)時(shí)加載全表把表達(dá)式編譯成內(nèi)存對(duì)象逐條對(duì)申請(qǐng)數(shù)據(jù)做匹配。action表示命中后干什么priority決定多條規(guī)則沖突時(shí)聽誰的。把規(guī)則從代碼里抽出來放到表里最大的價(jià)值是風(fēng)控業(yè)務(wù)人員可以直接配置、即時(shí)生效不需要等開發(fā)排期上線。模型打分這部分選型的核心矛盾是跨語言部署和性能。常見做法有三種把模型導(dǎo)出為PMML格式用Java加載打分導(dǎo)出為ONNX格式用推理引擎加載或者直接用Spark ML庫(kù)的Pipeline保存和加載。PMML在信貸風(fēng)控里用得仍然很廣因?yàn)樗鼘?duì)特征順序敏感字段名一一對(duì)應(yīng)部署方不需要關(guān)心訓(xùn)練方的代碼細(xì)節(jié)。用Spark訓(xùn)練完的模型保存并加載做打分流程最順但需要保證運(yùn)行環(huán)境里有匹配的Spark版本依賴。3.4 任務(wù)調(diào)度與血緣離線任務(wù)為什么必須配調(diào)度信貸系統(tǒng)的離線任務(wù)數(shù)量通常幾十上百個(gè)靠crontab或人工觸發(fā)根本管理不過來。源碼包里如果有部署文檔調(diào)度配置一定是重點(diǎn)章節(jié)。調(diào)度平臺(tái)要做的事是兩件一是按依賴關(guān)系編排任務(wù)上游產(chǎn)出分區(qū)數(shù)據(jù)后下游才能啟動(dòng)二是失敗重試和告警任務(wù)掛了自動(dòng)重試幾次再不行就通知值班的人。任務(wù)依賴編排的核心是數(shù)據(jù)就緒檢測(cè)。一個(gè)Spark任務(wù)啟動(dòng)前要先確認(rèn)它依賴的Hive分區(qū)已經(jīng)生成而不是用sleep硬等。常見做法是調(diào)度平臺(tái)支持前置檢查節(jié)點(diǎn)檢查SQL返回的分區(qū)數(shù)據(jù)量是否大于0。離線跑批鏈路里ODS同步?jīng)Q定了今天有沒有原材料DWD清洗決定了加工質(zhì)量DWS特征寬表決定了模型和報(bào)表能否產(chǎn)出。任何一環(huán)掛掉后面的任務(wù)即使跑成功產(chǎn)出的也是臟數(shù)據(jù)。源碼里這一塊的工程價(jià)值最容易被低估但生產(chǎn)事故往往就出在這——任務(wù)報(bào)SUCCESS但數(shù)據(jù)是空的比任務(wù)失敗更難排查。血緣管理同樣值得提。每個(gè)特征字段要能追溯到它來自哪張?jiān)幢怼⒔?jīng)過了哪些加工步驟否則排查問題時(shí)只能靠翻代碼。實(shí)際項(xiàng)目里我會(huì)要求所有特征加工腳本都登記一張?jiān)幢?目標(biāo)表映射表調(diào)度平臺(tái)跑完任務(wù)后就更新血緣信息半年下來這套元數(shù)據(jù)比代碼注釋可靠得多。4. 讓源碼落地集群規(guī)劃、打包提交與必調(diào)參數(shù)4.1 環(huán)境選型物理機(jī)、云主機(jī)還是容器化的判斷標(biāo)準(zhǔn)拿到源碼后的第一個(gè)決定是在什么環(huán)境上跑。判斷標(biāo)準(zhǔn)其實(shí)就三條數(shù)據(jù)量、任務(wù)節(jié)奏、運(yùn)維能力。數(shù)據(jù)量在幾TB到幾十TB、任務(wù)以T1批處理為主的用幾臺(tái)物理機(jī)構(gòu)建小型Hadoop集群最經(jīng)濟(jì)。數(shù)據(jù)量在百TB級(jí)、還需要彈性擴(kuò)容應(yīng)對(duì)大促高峰的云主機(jī)會(huì)更合適按需擴(kuò)容、用完釋放。想跑在容器里的也可以但Hadoop生態(tài)的容器化方案對(duì)網(wǎng)絡(luò)和存儲(chǔ)要求比較高運(yùn)維能力不夠的話不建議一步到位。規(guī)模參考按節(jié)點(diǎn)數(shù)給個(gè)常見范圍5臺(tái)以內(nèi)屬于入門規(guī)模適合跑通源碼和開發(fā)環(huán)境10到20臺(tái)能支撐千萬級(jí)用戶的風(fēng)控跑批再往上就要考慮業(yè)務(wù)拆分了。機(jī)器規(guī)格方面NameNode節(jié)點(diǎn)對(duì)內(nèi)存敏感建議64GB起步DataNode節(jié)點(diǎn)磁盤容量?jī)?yōu)先計(jì)算密集時(shí)CPU核數(shù)多給一些。Spark的executor和HDFS的DataNode通常混部在同一批節(jié)點(diǎn)上這樣數(shù)據(jù)本地性最好——Spark讀數(shù)據(jù)時(shí)直接從本機(jī)磁盤拿省去網(wǎng)絡(luò)傳輸。4.2 從源碼到運(yùn)行打包、上傳、提交 Spark 作業(yè)源碼工程一般是Maven或Gradle組織的多模塊工程拿到后先編譯打包再提交到集群運(yùn)行。打包和提交的命令如下mvn clean package -DskipTests -Pdist hdfs dfs -mkdir -p /apps/credit-risk hdfs dfs -put credit-risk-1.0.jar /apps/credit-risk/ spark-submit \ --master yarn \ --deploy-mode cluster \ --class com.example.credit.CreditBatchMain \ --files hdfs:///apps/credit/risk_rule.conf \ --conf spark.executor.extraJavaOptions-Dconfig.filerisk_rule.conf \ --queue risk \ --executor-memory 16g \ --executor-cores 4 \ --num-executors 20 \ /apps/credit-risk/credit-risk-1.0.jar \ --env prod \ --dt 2025-01-15各參數(shù)作用拆開看--master yarn表示交給集群資源調(diào)度--deploy-mode cluster讓Driver也跑在集群里適合離線批任務(wù)--class指定要運(yùn)行的入口類--files把外部配置文件分發(fā)到每個(gè)執(zhí)行節(jié)點(diǎn)--queue指定YARN隊(duì)列把風(fēng)控任務(wù)單獨(dú)放一個(gè)隊(duì)列可以避免和其他業(yè)務(wù)互相搶占資源最后兩個(gè)參數(shù)--env和--dt是業(yè)務(wù)參數(shù)--dt就是第2章里提到的業(yè)務(wù)日期特征計(jì)算任務(wù)都依賴它來定位分區(qū)。初次啟動(dòng)時(shí)建議先不用--num-executors改用Spark的動(dòng)態(tài)資源分配讓集群按負(fù)載自動(dòng)伸縮。等穩(wěn)定運(yùn)行一段時(shí)間后再根據(jù)監(jiān)控?cái)?shù)據(jù)固定executor數(shù)量或直接保留動(dòng)態(tài)策略。源碼里如果自帶啟動(dòng)腳本重點(diǎn)看腳本里的參數(shù)值——不同公司集群規(guī)模差異很大直接照搬的后果通常不是OOM就是資源浪費(fèi)。4.3 必調(diào)參數(shù)executor 內(nèi)存、分區(qū)數(shù)與動(dòng)態(tài)資源Spark作業(yè)調(diào)參是玄學(xué)但有幾條基本邏輯不是玄學(xué)。Executor內(nèi)存、CPU核數(shù)、分區(qū)數(shù)三者必須放在一起看單獨(dú)調(diào)哪一個(gè)都可能翻車。常見參數(shù)配置參考如下參數(shù)常見取值調(diào)整依據(jù)spark.executor.memory8g ~ 32g每個(gè)任務(wù)的數(shù)據(jù)處理量、是否有大聚合spark.executor.cores2 ~ 4越高并行度越高但要控制單節(jié)點(diǎn)總核數(shù)spark.sql.shuffle.partitions200 ~ 2000和shuffle數(shù)據(jù)量掛鉤過少導(dǎo)致單分區(qū)壓力過大過多導(dǎo)致小文件膨脹spark.dynamicAllocation.enabledtrue / false生產(chǎn)環(huán)境建議開啟讓集群彈性伸縮spark.yarn.executor.memoryOverhead1g ~ 4g存放JVM堆外內(nèi)存頻繁O(jiān)OM時(shí)優(yōu)先調(diào)這里spark.sql.adaptive.enabledtrue動(dòng)態(tài)調(diào)整reduce分區(qū)數(shù)解決數(shù)據(jù)傾斜第一次跑源碼任務(wù)時(shí)我一般先把shuffle分區(qū)分母設(shè)成executor總核數(shù)的2到3倍比如20個(gè)executor、每個(gè)4核、共80核分區(qū)數(shù)設(shè)160到240然后看Spark UI里每個(gè)stage的耗時(shí)分布。如果某個(gè)stage最后幾個(gè)task明顯比別的慢很多大概率是數(shù)據(jù)傾斜。解決數(shù)據(jù)傾斜優(yōu)先開spark.sql.adaptive.enabled還不行再考慮加鹽或者拆廣播變量。executor內(nèi)存溢出則優(yōu)先調(diào)memoryOverhead它不是堆內(nèi)存但堆外內(nèi)存不夠一樣會(huì)OOM。還有一個(gè)容易被忽略的參數(shù)是spark.speculation開啟后Spark會(huì)重跑那些明顯滯后的task取最先完成的那個(gè)結(jié)果。對(duì)大數(shù)據(jù)量的批處理任務(wù)能兜底但也可能帶來重復(fù)計(jì)算穩(wěn)妥做法是在跑的慢的節(jié)點(diǎn)上臨時(shí)開啟穩(wěn)定后關(guān)掉。參數(shù)調(diào)整后必須觀察Spark UI上的GC時(shí)間、shuffle落地量、單task處理時(shí)長(zhǎng)三個(gè)指標(biāo)不要只看任務(wù)成功還是失敗。5. 風(fēng)控系統(tǒng)落地避坑五條值得記住的排查記錄5.1 現(xiàn)象夜間批處理任務(wù)堆積Spark 作業(yè)反復(fù)失敗某次生產(chǎn)環(huán)境凌晨的T1批處理任務(wù)在3點(diǎn)后開始堆積Spark作業(yè)提交成功但運(yùn)行幾分鐘后就被YARN殺掉?，F(xiàn)象是任務(wù)反復(fù)失敗日志里出現(xiàn)大量ExecutorLostFailure和Container killed messages。原因是executor申請(qǐng)的內(nèi)存超過節(jié)點(diǎn)可用內(nèi)存加上多個(gè)任務(wù)在同一批節(jié)點(diǎn)上同時(shí)運(yùn)行物理內(nèi)存被打滿后觸發(fā)系統(tǒng)OOM容器直接被回收。這不是單次任務(wù)的問題而是所有夜間批任務(wù)擠在同一時(shí)間窗口、資源池沒有隔離的必然結(jié)果。解決分兩步先給風(fēng)控任務(wù)單獨(dú)劃分YARN隊(duì)列限制隊(duì)列最大資源占用防止批任務(wù)之間互相爭(zhēng)搶再調(diào)小單executor內(nèi)存把16g降到8g同時(shí)增加executor數(shù)量保住總計(jì)算能力。更根本的辦法是把大任務(wù)拆成多個(gè)有依賴關(guān)系的小任務(wù)錯(cuò)峰執(zhí)行讓峰值內(nèi)存需求錯(cuò)開。5.2 現(xiàn)象實(shí)時(shí)評(píng)分接口延遲抖動(dòng)從 200ms 漲到 2s風(fēng)控實(shí)時(shí)評(píng)分接口平時(shí)穩(wěn)定在200ms左右某段時(shí)間開始下午高峰期漲到2s甚至超時(shí)?，F(xiàn)象沒有規(guī)律不是每次都慢但慢的時(shí)候集中在流量大的時(shí)間段。抓線程和GC日志后發(fā)現(xiàn)在線評(píng)分服務(wù)讀取特征緩存時(shí)每次都用同步方式請(qǐng)求遠(yuǎn)程緩存一旦緩存服務(wù)出現(xiàn)網(wǎng)絡(luò)抖動(dòng)或者慢查詢調(diào)用方線程全部阻塞在等待上。另外服務(wù)本身的JVM堆設(shè)置偏小高峰期Full GC頻繁也會(huì)放大延遲。解決是在評(píng)分服務(wù)本地加一層二級(jí)緩存把熱點(diǎn)特征緩存到本地內(nèi)存設(shè)置2分鐘的過期時(shí)間遠(yuǎn)程緩存只在本地未命中時(shí)回源。同時(shí)把遠(yuǎn)程緩存查詢改成異步刷新請(qǐng)求進(jìn)來先查本地緩存查不到就直接回源一次后臺(tái)再異步更新。改動(dòng)后接口延遲穩(wěn)定在100ms到300ms之間抖動(dòng)問題沒有再出現(xiàn)。5.3 現(xiàn)象訓(xùn)練集和線上特征口徑對(duì)不上模型 AUC 虛高模型在回測(cè)時(shí)AUC到了0.82上線后實(shí)際區(qū)分能力明顯達(dá)不到回測(cè)水平調(diào)閾值也沒用。排查后發(fā)現(xiàn)訓(xùn)練特征和上線特征不是一個(gè)源頭。根因在特征計(jì)算任務(wù)的日期參數(shù)上。訓(xùn)練樣本的特征是歷史某一天跑的批任務(wù)產(chǎn)出的當(dāng)時(shí)用的窗口終點(diǎn)是當(dāng)天線上評(píng)分時(shí)特征服務(wù)用的窗口終點(diǎn)是當(dāng)前時(shí)刻。對(duì)同一個(gè)申請(qǐng)單如果特征引用的是未來一段時(shí)間的流水訓(xùn)練出來的特征分布和線上完全不同。更隱蔽的是缺失值處理邏輯不一致訓(xùn)練時(shí)用了全表均值填充線上實(shí)時(shí)服務(wù)沒有這個(gè)均值表填的0分布直接偏移。解決是整個(gè)團(tuán)隊(duì)統(tǒng)一用特征生成腳本的同一份代碼訓(xùn)練和線上共用一套特征邏輯。新增一個(gè)特征校驗(yàn)任務(wù)每天對(duì)比訓(xùn)練特征表和線上特征表的同名字段的均值、標(biāo)準(zhǔn)差、分布差異PSI超閾值就報(bào)警。這個(gè)機(jī)制上線后口徑不一致的問題基本做到了早發(fā)現(xiàn)早處理。5.4 現(xiàn)象Hive 小文件暴漲NameNode 內(nèi)存吃緊集群NameNode堆內(nèi)存使用率持續(xù)攀升Hive表查詢?cè)絹碓铰２榭春蟀l(fā)現(xiàn)DWS層特征表的文件數(shù)量從上萬個(gè)漲到了十幾萬個(gè)大量幾KB到幾百KB的小文件占滿了NameNode內(nèi)存。原因是特征計(jì)算任務(wù)用了動(dòng)態(tài)分區(qū)寫Hive同時(shí)shuffle分區(qū)數(shù)設(shè)得過大Spark寫出的文件數(shù)量等于shuffle分區(qū)數(shù)乘以動(dòng)態(tài)分區(qū)數(shù)數(shù)據(jù)量沒漲文件數(shù)漲了好幾倍。小文件多不僅拖垮NameNode還會(huì)讓后續(xù)讀取任務(wù)啟動(dòng)大量task每個(gè)task都在做元數(shù)據(jù)操作整個(gè)集群變慢。解決分兩層Spark寫入前先做一次repartition把輸出分區(qū)數(shù)控制在合理范圍內(nèi)寫入后跑一次合并任務(wù)用Hive的MERGE或Spark repartition按新分區(qū)數(shù)重寫一遍。參數(shù)方面把spark.sql.shuffle.partitions從2000降到400匹配實(shí)際數(shù)據(jù)規(guī)模而不是簡(jiǎn)單照搬。關(guān)鍵是建立一條小文件巡檢任務(wù)每天統(tǒng)計(jì)各層表的文件數(shù)變化超過閾值就自動(dòng)觸發(fā)合并。5.5 現(xiàn)象模型上線后壞賬率不降反升新模型上線一個(gè)月后審批通過率沒有明顯變化壞賬率卻比舊模型時(shí)期高了一截。團(tuán)隊(duì)第一反應(yīng)是模型不行但要區(qū)分是模型本身的問題還是客群結(jié)構(gòu)變化的問題。排查后發(fā)現(xiàn)新模型確實(shí)更準(zhǔn)但它會(huì)把一批舊模型認(rèn)為高風(fēng)險(xiǎn)、拒絕掉的人群放進(jìn)來。這批人其實(shí)是舊模型時(shí)期被拒絕、沒有真實(shí)貸后表現(xiàn)、只能靠拒絕推斷去估計(jì)其風(fēng)險(xiǎn)的人。新模型使用了更寬泛的特征給了部分原本該拒絕的人通過而這些人的實(shí)際表現(xiàn)比訓(xùn)練時(shí)預(yù)估的更差。解決思路是引入拒絕推斷的修正樣本把被拒絕人群的風(fēng)險(xiǎn)標(biāo)簽做權(quán)重調(diào)整后放進(jìn)訓(xùn)練集同時(shí)上線前做嚴(yán)格的回測(cè)和影子評(píng)分。更穩(wěn)妥的做法是新模型分階段放量先放行新模型判定為低風(fēng)險(xiǎn)的申請(qǐng)觀察幾周真實(shí)貸后表現(xiàn)再逐步放開閾值。這樣即使模型有偏差損失也可控。6. 拿來即用的進(jìn)階驗(yàn)證回測(cè)與影子評(píng)分給上線前的模型兜底回測(cè)和影子評(píng)分是我在信貸風(fēng)控項(xiàng)目里最看重的兩道驗(yàn)證工序?；販y(cè)的核心是時(shí)間切分把歷史數(shù)據(jù)按時(shí)間切兩段前一段用來訓(xùn)練后一段用來驗(yàn)證。和一般機(jī)器學(xué)習(xí)隨機(jī)切分不同風(fēng)控回測(cè)必須按時(shí)間切因?yàn)樾刨J樣本天然有時(shí)序關(guān)系隨機(jī)切分會(huì)把未來信息泄漏進(jìn)訓(xùn)練集。切分點(diǎn)一般選在表現(xiàn)期之前比如驗(yàn)證段要留出90天表現(xiàn)期那就至少把最后90天的申請(qǐng)排除出訓(xùn)練集?；販y(cè)指標(biāo)要同時(shí)看區(qū)分度、排序穩(wěn)定性和群體穩(wěn)定性三類區(qū)分度看KS和AUC排序穩(wěn)定性看每個(gè)分?jǐn)?shù)段通過率是否和訓(xùn)練期一致群體穩(wěn)定性看PSIPSI超過0.25說明推理時(shí)的特征分布和訓(xùn)練期差異已經(jīng)很大模型需要重訓(xùn)。影子評(píng)分是把新模型旁路部署在線上但不參與實(shí)際決策生產(chǎn)請(qǐng)求進(jìn)來舊模型打分決定審批結(jié)果新模型同時(shí)打分但結(jié)果只落庫(kù)不生效。跑上一到兩個(gè)星期拿新模型的分和舊模型的分做對(duì)比看排序一致性、各分?jǐn)?shù)段通過率差異、特征分布偏移情況。影子評(píng)分能驗(yàn)證一件事回測(cè)驗(yàn)證不了的事特征在真實(shí)線上環(huán)境里取數(shù)是否穩(wěn)定。我在一個(gè)項(xiàng)目上吃過虧回測(cè)時(shí)AUC很漂亮影子跑了一周發(fā)現(xiàn)新模型的評(píng)分均值比舊模型低了將近30分原因是線上特征服務(wù)對(duì)某些長(zhǎng)尾特征取到的值大量為0和訓(xùn)練時(shí)的分布完全脫節(jié)。從那次以后我養(yǎng)成一個(gè)習(xí)慣任何模型上線前必須跑至少一周影子評(píng)分并對(duì)比分位數(shù)、PSI和特征覆蓋率三個(gè)報(bào)表。這個(gè)習(xí)慣雖然讓上線周期拉長(zhǎng)了幾天但攔下了不少看著好看、落地就翻車的模型。這套驗(yàn)證鏈路值得每一個(gè)做信貸風(fēng)控的團(tuán)隊(duì)堅(jiān)持希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取