4. 批处理架构详解：批处理原理与MapReduce、Spark批处理引擎、批处理在实体解析中的适用场景

批处理（Batch Processing）是大数据时代的基石，也是金融图谱实体解析中处理历史数据、全量归约与离线校验的核心计算范式。本章深入剖析批处理原理，对比MapReduce与Spark两大引擎的架构差异，并明确批处理在实体解析中的最佳适用场景。

4.1 批处理核心原理

批处理遵循“数据不动代码动”或“数据分片并行处理”的原则，其核心特征包括：

• 全量数据扫描：一次性读取整个数据集（如HDFS上的历史交易快照）。
• 高吞吐量：通过数据本地化与并行任务，最大化磁盘I/O与CPU利用率。
• 延迟容忍：分钟级到小时级完成，适合非实时场景。
• 容错性：通过任务重试与数据副本机制保证最终一致性。

4.2 MapReduce 批处理引擎

MapReduce是Google提出的经典批处理模型，分为Map（映射）和Reduce（归约）两个阶段，中间通过Shuffle连接。在金融图谱实体解析中，MapReduce常用于大规模实体ID映射与属性聚合。

4.2.1 MapReduce 执行流程

1. Input Split：将输入文件切分为逻辑分片（Split），每个Split对应一个Map任务。
2. Map Phase：读取Split中的键值对，输出中间键值对（如实体ID → 属性列表）。
3. Shuffle & Sort：将相同Key的中间结果路由到同一Reduce节点，并按Key排序。
4. Reduce Phase：对每个Key的Value列表执行归约逻辑（如合并实体属性、去重）。
5. Output：将结果写入HDFS或外部存储。

4.2.2 MapReduce 与 Spark 对比

4.3 Spark 批处理引擎

Spark基于内存计算，通过RDD（Resilient Distributed Dataset）抽象实现高效的批处理。在实体解析中，Spark的DataFrame API与GraphX图计算库为实体对齐与关系挖掘提供了强大支持。

4.3.1 Spark 批处理核心组件

• RDD：不可变、可分区、可并行操作的数据集合，支持transformation（如map、filter）与action（如reduce、count）。
• DAG Scheduler：将RDD的血统图转换为阶段（Stage）与任务（Task），优化计算流水线。
• Shuffle：与MapReduce类似，但Spark通过Sort-Based Shuffle减少磁盘写入。
• Catalyst Optimizer：DataFrame/SQL的查询优化器，支持谓词下推、列剪枝等优化。

4.4 批处理在实体解析中的适用场景

并非所有实体解析任务都适合批处理。以下场景是批处理的“主场”，而实时流处理则更适合增量场景。

4.5 批处理架构总览图

下图展示了从数据源到实体解析结果的批处理全链路架构，涵盖MapReduce与Spark两种引擎的典型流程。

4.6 批处理在实体解析中的代码示例

以下Spark批处理代码展示了实体解析中常见的“实体属性归约”操作：对同一实体ID的多源属性进行合并与去重。

// Spark批处理：实体属性归约示例
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder()
  .appName("EntityResolutionBatch")
  .config("spark.sql.shuffle.partitions", "200")
  .getOrCreate()

// 读取多源实体数据（假设已加载为DataFrame）
val entityDF = spark.read.parquet("hdfs://data/entity_sources/")

// 实体属性归约：按实体ID聚合，合并姓名、证件号、地址，去重
val resolvedDF = entityDF
  .groupBy("entity_id")
  .agg(
    collect_set("name").alias("names"),
    collect_set("id_number").alias("id_numbers"),
    collect_set("address").alias("addresses"),
    count("*").alias("source_count")
  )
  .withColumn("resolved_name", when(size(col("names")) === 1, col("names")(0))
    .otherwise(concat_ws("|", col("names"))))
  .withColumn("resolved_id", when(size(col("id_numbers")) === 1, col("id_numbers")(0))
    .otherwise(array_join(col("id_numbers"), "|")))

// 写入结果
resolvedDF.write.mode("overwrite").parquet("hdfs://data/entity_resolved/")

spark.stop()

4.7 批处理 vs 流处理：实体解析场景选择