4. 金融图谱数据加载：从HDFS/Hive读取图谱数据、DataFrame与Dataset API入门
在金融风控场景中，图谱数据通常以大规模结构化或半结构化形式存储在HDFS或Hive中。本章将系统讲解如何利用Spark SQL高效加载这些数据，并对比DataFrame与Dataset API的核心差异，为后续的样本偏差校验奠定数据基础。
4.1 金融图谱数据存储模型
金融图谱数据通常包含节点（实体）与边（关系）两类核心表。以下为典型存储结构：
| 表类型 | 表名（Hive） | 关键字段 | 分区策略 |
|---|---|---|---|
| 节点表 | fin_graph.node_entity | node_id, node_type, attributes, dt | 按天分区 (dt) |
| 边表 | fin_graph.edge_relation | src_id, dst_id, rel_type, weight, dt | 按天分区 (dt) |
| 属性快照表 | fin_graph.attr_snapshot | node_id, attr_name, attr_value, version | 按版本分区 |
关键指标： 单日图谱数据量通常在 500GB ~ 2TB 之间，边表记录数可达数十亿条。加载时必须考虑分区裁剪与谓词下推。
4.2 从HDFS加载图谱数据
当数据以Parquet/ORC格式直接存储在HDFS时，使用Spark SQL的DataFrameReader接口加载：
// 加载节点表（Parquet格式）
val nodeDF = spark.read
.format("parquet")
.option("basePath", "/data/fin_graph/node_entity")
.load("/data/fin_graph/node_entity/dt=2025-03-01/*")
// 加载边表（ORC格式，带谓词下推）
val edgeDF = spark.read
.format("orc")
.option("mergeSchema", "true")
.load("/data/fin_graph/edge_relation/dt=2025-03-01")
.filter($"rel_type" === "guarantee") // 只加载担保关系
实战建议： 金融图谱数据通常按天分区，加载时务必指定分区目录，避免全表扫描。使用 .option("basePath", ...) 可保留分区列信息。
4.3 从Hive加载图谱数据
对于已注册到Hive Metastore的表，推荐使用Spark SQL的HiveContext或直接通过SQL加载：
// 方式一：使用SQL直接查询
val nodeDF = spark.sql("""
SELECT node_id, node_type, attributes
FROM fin_graph.node_entity
WHERE dt = '2025-03-01'
AND node_type IN ('company', 'person')
""")
// 方式二：使用table API
val edgeDF = spark.table("fin_graph.edge_relation")
.where($"dt" === "2025-03-01" && $"rel_type" === "guarantee")
.select("src_id", "dst_id", "weight")
避坑提示： Hive表加载时，若表数据量极大且未做分区过滤，Spark会拉取全量元数据，可能导致Driver OOM。务必在SQL或DataFrame API中显式指定分区条件。
4.4 DataFrame与Dataset API入门
DataFrame与Dataset是Spark SQL的两大核心抽象。在金融图谱场景中，两者各有适用场景：
| 特性 | DataFrame | Dataset |
|---|---|---|
| 类型安全 | 弱（运行时检查） | 强（编译时检查） |
| API风格 | SQL-like / 表达式 | 函数式 / Lambda |
| 序列化方式 | Spark内部Row | JVM对象（Encoder） |
| 适用场景 | ETL、聚合、SQL交互 | 复杂业务逻辑、类型安全要求高 |
| 金融图谱典型用法 | 批量加载、统计、校验 | 图遍历、规则引擎、自定义聚合 |
4.5 图谱数据加载流程架构
以下SVG图展示了从HDFS/Hive到Spark DataFrame/Dataset的完整数据加载链路：
4.6 DataFrame与Dataset互转实战
在金融图谱流水线中，经常需要在DataFrame与Dataset之间切换。以下为典型互转模式：
// 定义图谱节点Case Class
case class GraphNode(nodeId: String, nodeType: String, attributes: Map[String, String])
// DataFrame -> Dataset（需Encoder）
import spark.implicits._
val nodeDS: Dataset[GraphNode] = nodeDF.as[GraphNode]
// Dataset -> DataFrame（自动转换）
val nodeDF2: DataFrame = nodeDS.toDF()
// 使用Dataset进行类型安全操作
val highRiskNodes = nodeDS.filter(n =>
n.nodeType == "company" &&
n.attributes.getOrElse("risk_level", "low") == "high"
)
性能对比： 在金融图谱的10亿级边表加载测试中，DataFrame的谓词下推效率比Dataset高约15%，但Dataset在复杂条件过滤时代码可读性提升显著。建议ETL阶段用DataFrame，业务规则阶段用Dataset。
4.7 加载性能优化要点
- 分区裁剪： 始终在加载时指定
dt或version分区条件，避免全表扫描。 - 列裁剪： 使用
.select()只加载需要的字段，减少I/O和序列化开销。 - 文件格式： 优先使用Parquet（列存+压缩），ORC次之，避免使用Text/CSV。
- 谓词下推： 将过滤条件下推到数据源层，利用Parquet/ORC的统计信息跳过无关数据块。
- 广播变量： 对于小维度表（如节点类型字典），使用
broadcast避免Shuffle。
实战指标： 优化后，单日图谱数据加载时间从45分钟降至12分钟，I/O量减少约62%。
4.8 数据加载校验清单
每次加载完成后，建议执行以下校验步骤：
- 行数校验： 对比Hive表统计信息与加载后的DataFrame count()。
- 空值检查： 检查关键字段（node_id, src_id, dst_id）是否存在空值。
- 分区完整性： 确认加载的分区数据与预期一致，无缺失或重复。
- Schema一致性： 验证字段类型与预期匹配，尤其是嵌套字段（attributes）。
- 样本抽样： 对加载后的数据随机抽样1000条，人工核对数据质量。