您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

TF-IDF MapReduce 實現全流程：三階段作業鏈、IK 分詞、分布式緩存優化與精準營銷應用

編輯：Java 相關知識

基於 Hadoop MapReduce 實現大規模文本語料的 TF-IDF 計算，採用三階段作業鏈設計，整合 IK 分詞器、分布式緩存優化、自定義分區器，最終支撐電商精準營銷推薦系統。

一、TF-IDF 核心原理回顧

TF (Term Frequency)：詞 w 在文檔 d 中出現頻率，TF(w,d) = count(w,d) / |d|
IDF (Inverse Document Frequency)：詞 w 的區分度，IDF(w) = log(N / (DF(w) + 1))，N=總文檔數，DF=包含 w 的文檔數
TF-IDF：TF(w,d) × IDF(w)，值越大表示詞對文檔越具代表性

二、數據集與預處理


// 原始數據格式：TSV (微博ID \t 內容)
3823890314914825	今天天氣晴好，姐妹們約起，一起去逛街。
3823890314914826	剛買了新款iPhone，拍照真不錯！
...

三、三階段 MapReduce 作業鏈設計

階段一：分詞 + 計算 TF + 統計總文檔數


// Mapper
public static class TFMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private IKSegmenter segmenter; // IK 分詞器
    
    @Override
    protected void setup(Context context) {
        segmenter = new IKSegmenter(true); // 智能分詞模式
    }
    
    @Override
    protected void map(LongWritable key, Text value, Context context) {
        String[] fields = value.toString().split("\t");
        if (fields.length < 2) return;
        
        String docId = fields[0];
        String content = fields[1];
        
        // 1. 分詞
        List words = segmenter.segment(content);
        
        // 2. 輸出 (word_docId, 1) 用於 TF 計數
        for (String word : words) {
            String compositeKey = word + "_" + docId;
            context.write(new Text(compositeKey), new IntWritable(1));
        }
        
        // 3. 輸出總文檔數計數 (特殊 Key: "DOC_COUNT")
        context.write(new Text("DOC_COUNT"), new IntWritable(1));
    }
}

// 自定義分區器：DOC_COUNT 路由到最後一個 Reduce
public class TFPartitioner extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text key, IntWritable value, int numPartitions) {
        return "DOC_COUNT".equals(key.toString()) ? numPartitions - 1 : 
               (key.hashCode() & Integer.MAX_VALUE) % (numPartitions - 1);
    }
}

// Reducer
public static class TFReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) {
        int sum = 0;
        for (IntWritable v : values) sum += v.get();
        context.write(key, new IntWritable(sum));
    }
}

// 輸出示例 (4個分區文件，最後一個包含 DOC_COUNT):
// part-r-00000: 今天_3823890314914825	2
// part-r-00001: 天氣_3823890314914825	1
// part-r-00003: DOC_COUNT	1075

階段二：計算 DF (Document Frequency)


// Mapper: 讀取階段一輸出，排除 DOC_COUNT 文件
public static class DFMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    
    @Override
    protected void map(LongWritable key, Text value, Context context) {
        // 獲取輸入文件名，跳過包含 DOC_COUNT 的最後一個分區文件
        FileSplit split = (FileSplit) context.getInputSplit();
        String fileName = split.getPath().getName();
        if (fileName.endsWith("part-r-00003")) return; // 假設4個Reduce，最後一個索引3
        
        String[] parts = value.toString().split("\t");
        if (parts.length < 2) return;
        
        String compositeKey = parts[0]; // word_docId
        String[] kv = compositeKey.split("_");
        if (kv.length < 2) return;
        
        String word = kv[0];
        // 輸出 (word, 1) - 每個文檔對該詞貢獻 1
        context.write(new Text(word), new IntWritable(1));
    }
}

// Reducer: 統計每個詞出現在多少文檔中
public static class DFReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) {
        int df = 0;
        for (IntWritable v : values) df += v.get();
        context.write(key, new IntWritable(df));
    }
}

// 輸出示例:
// 今天	5
// 天氣	3
// 逛街	2

階段三：計算 TF-IDF 並聚合輸出


// Mapper: 讀取階段一 TF 結果，利用分布式緩存加載 DOC_COUNT 和 DF 表
public static class TFIDFMapper extends Mapper<LongWritable, Text, Text, Text> {
    private long totalDocs; // N
    private Map<String, Integer> dfMap = new HashMap<>();
    
    @Override
    protected void setup(Context context) throws IOException {
        // 1. 加載總文檔數 (DOC_COUNT)
        URI[] cacheFiles = context.getCacheFiles();
        for (URI uri : cacheFiles) {
            Path path = new Path(uri);
            String fileName = path.getName();
            try (BufferedReader br = new BufferedReader(
                    new InputStreamReader(FileSystem.getLocal(context.getConfiguration()).open(path)))) {
                String line;
                while ((line = br.readLine()) != null) {
                    String[] parts = line.split("\t");
                    if (fileName.contains("part-r-00003") && parts[0].equals("DOC_COUNT")) {
                        totalDocs = Long.parseLong(parts[1]);
                    } else if (!fileName.contains("part-r-00003")) { // DF 文件
                        if (parts.length == 2) {
                            dfMap.put(parts[0], Integer.parseInt(parts[1]));
                        }
                    }
                }
            }
        }
    }
    
    @Override
    protected void map(LongWritable key, Text value, Context context) {
        String[] parts = value.toString().split("\t");
        if (parts.length < 2) return;
        
        String compositeKey = parts[0]; // word_docId
        int tf = Integer.parseInt(parts[1]);
        
        String[] kv = compositeKey.split("_");
        if (kv.length < 2) return;
        String word = kv[0];
        String docId = kv[1];
        
        Integer df = dfMap.get(word);
        if (df == null || df == 0) return; // 防守
        
        // TF-IDF = tf * log(N / df)
        double tfidf = tf * Math.log((double) totalDocs / df);
        
        // 輸出 (docId, word:tfidf)
        context.write(new Text(docId), new Text(word + ":" + tfidf));
    }
}

// Reducer: 聚合同一文檔的所有詞 TF-IDF
public static class TFIDFReducer extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) {
        StringBuilder sb = new StringBuilder();
        boolean first = true;
        for (Text val : values) {
            if (!first) sb.append(",");
            sb.append(val.toString());
            first = false;
        }
        context.write(key, new Text(sb.toString()));
    }
}

// 最終輸出示例:
// 3823890314914825	今天:2.788,逛街:3.981,姐妹:1.987
// 3823890314914826	iPhone:4.123,拍照:2.456,新款:1.789

四、Job 驅動與分布式緩存配置


public class TFIDFDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        
        // === Job 1: TF + DOC_COUNT ===
        Job job1 = Job.getInstance(conf, "TF-IDF Stage1: TF & DocCount");
        job1.setJarByClass(TFIDFDriver.class);
        job1.setMapperClass(TFMapper.class);
        job1.setPartitionerClass(TFPartitioner.class);
        job1.setReducerClass(TFReducer.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(IntWritable.class);
        job1.setNumReduceTasks(4); // 關鍵：4個Reduce，最後一個處理DOC_COUNT
        FileInputFormat.addInputPath(job1, new Path(args[0]));
        FileOutputFormat.setOutputPath(job1, new Path("/user/tfidf/output/stage1"));
        boolean success1 = job1.waitForCompletion(true);
        
        if (!success1) return;
        
        // === Job 2: DF ===
        Job job2 = Job.getInstance(conf, "TF-IDF Stage2: DF");
        job2.setJarByClass(TFIDFDriver.class);
        job2.setMapperClass(DFMapper.class);
        job2.setReducerClass(DFReducer.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job2, new Path("/user/tfidf/output/stage1"));
        FileOutputFormat.setOutputPath(job2, new Path("/user/tfidf/output/stage2"));
        boolean success2 = job2.waitForCompletion(true);
        
        if (!success2) return;
        
        // === Job 3: TF-IDF ===
        Job job3 = Job.getInstance(conf, "TF-IDF Stage3: TF-IDF Calc");
        job3.setJarByClass(TFIDFDriver.class);
        job3.setMapperClass(TFIDFMapper.class);
        job3.setReducerClass(TFIDFReducer.class);
        job3.setOutputKeyClass(Text.class);
        job3.setOutputValueClass(Text.class);
        
        // 關鍵：分布式緩存加載 DOC_COUNT 和 DF 文件
        job3.addCacheFile(new Path("/user/tfidf/output/stage1/part-r-00003").toUri()); // DOC_COUNT
        job3.addCacheFile(new Path("/user/tfidf/output/stage2/part-r-00000").toUri()); // DF
        
        FileInputFormat.addInputPath(job3, new Path("/user/tfidf/output/stage1"));
        FileOutputFormat.setOutputPath(job3, new Path("/user/tfidf/output/stage3"));
        job3.waitForCompletion(true);
    }
}

五、精準營銷應用場景

5.1 用戶畫像構建


// 每個用戶(微博)獲得 Top-K 高 TF-IDF 詞作為興趣標籤
// 3823890314914825 → [逛街:3.98, 今天:2.79, 姐妹:1.99]
// 3823890314914826 → [iPhone:4.12, 拍照:2.46, 新款:1.79]

5.2 商家推薦邏輯


// 韓式料理店推廣「大骨湯」
// 1. 計算「大骨湯」在所有用戶中的 TF-IDF
// 2. 篩選 TF-IDF 排名 Top 3 且包含「大骨湯」的用戶
// 3. 這批用戶即為精準推送目標群體
// 4. 可結合地理位置(LBS)、活躍度、消費能力等多維度再過濾

5.3 擴展：Item-Based 協同過濾基礎

TF-IDF 向量可作為文檔特徵向量
計算文檔間餘弦相似度，構建「相關微博」推薦
結合 Word2Vec/BERT 進一步語義增強

六、性能優化關鍵點總結

優化點	實現方式	收益
分詞器復用	Mapper setup() 初始化，map() 復用	避免每行創建分詞器對象
自定義分區器	DOC_COUNT 隔離到單獨 Reduce	避免數據傾斜，保證全局計數準確
分布式緩存	Job.addCacheFile() 加載小表到內存	避免 HDFS 重複讀取，Map 端 Join 消除 Shuffle
Combiner	TF/DF 階段啟用 Combiner (同 Reducer)	減少 Map 輸出數據量，降低網絡/磁盤 IO
壓縮	mapreduce.output.fileoutputformat.compress=true	減少中間結果存儲與傳輸開銷
數據傾斜處理	高頻詞(停用詞)過濾、自定義分區、傾斜 Key 拆分	解決 Reduce 端熱點

七、遷移建議：從 MapReduce 到 Spark/Flink


// Spark 版本核心邏輯 (極大簡化)
val docs = sc.textFile("hdfs://data/weibo.tsv")
  .map(line => line.split("\t"))
  .filter(arr => arr.length >= 2)
  .map(arr => (arr(0), IKSegmenter.segment(arr(1)))) // (docId, words)

// TF 計算
val tf = docs.flatMap { case (docId, words) => 
  words.groupBy(identity).mapValues(_.size).map { case (w, c) => ((w, docId), c) }
}

// DF 計算
val df = tf.map { case ((w, docId), c) => (w, 1) }.reduceByKey(_ + _)

// TF-IDF
val tfidf = tf.join(df.broadcast) // Map 端 Join
  .map { case (w, ((docId, tf), df)) => (docId, (w, tf * math.log(N.toDouble / df))) }
  .groupByKey()
  .mapValues(_.toList.sortBy(-_._2).take(20)) // Top-20

核心優勢：Spark 內存計算消除 3 個 Job 的磁盤落地與 Shuffle，DF Broadcast 變量替代分布式緩存，代碼從 300+ 行縮減至 30 行，迭代開發效率提升 10x+。