2. 环境搭建：Spark集群安装与配置、GraphX库引入、Neo4j与Spark集成、开发环境验证

本章是《Spark回放反欺诈图谱实战案例集》的基石环节。一个稳定、高性能的分布式计算环境，是后续所有反欺诈图谱算法与回放分析能够正确运行的前提。我们将按照生产级标准，逐步完成Spark集群的安装、GraphX图计算库的引入、Neo4j图数据库与Spark的集成，并最终通过一个简单的图计算任务验证整个开发环境。

2.1 环境概览与版本选型

在开始搭建之前，需要明确各组件的版本兼容性。错误的版本组合会导致集成失败或运行时异常。以下是我们本次实战采用的推荐版本组合：

2.2 Spark集群安装与配置

我们以3台服务器（master、worker1、worker2）为例，搭建Spark Standalone集群。以下步骤在所有节点上执行。

2.2.1 前置条件与下载

1. 确保所有节点已安装JDK 11，并配置JAVA_HOME环境变量。
2. 在所有节点上配置SSH免密登录，确保master节点可以无密码访问所有worker节点。
3. 下载Spark安装包并解压至统一目录（如/opt/spark）。

# 在master节点执行
wget https://archive.apache.org/dist/spark/spark-3.4.0/spark-3.4.0-bin-hadoop3.tgz
tar -xzf spark-3.4.0-bin-hadoop3.tgz -C /opt/
ln -s /opt/spark-3.4.0-bin-hadoop3 /opt/spark

2.2.2 配置文件修改

进入/opt/spark/conf目录，基于模板文件创建配置文件。

cp spark-env.sh.template spark-env.sh
cp workers.template workers
cp spark-defaults.conf.template spark-defaults.conf

spark-env.sh 核心配置：

export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64
export SPARK_MASTER_HOST=192.168.1.100   # master节点IP
export SPARK_MASTER_PORT=7077
export SPARK_WORKER_CORES=4
export SPARK_WORKER_MEMORY=8g
export SPARK_DAEMON_MEMORY=2g
export SPARK_HOME=/opt/spark
export PATH=$PATH:$SPARK_HOME/bin

workers 文件：列出所有worker节点的主机名或IP。

worker1
worker2

spark-defaults.conf 关键参数：

spark.master                     spark://192.168.1.100:7077
spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.kryoserializer.buffer.max  256m
spark.sql.adaptive.enabled       true
spark.sql.adaptive.coalescePartitions.enabled true

2.2.3 分发配置与启动集群

将配置好的Spark目录通过scp分发到所有worker节点，确保路径一致。

scp -r /opt/spark worker1:/opt/
scp -r /opt/spark worker2:/opt/

在master节点启动集群：

# 启动master
/opt/spark/sbin/start-master.sh

# 在所有worker节点启动worker
/opt/spark/sbin/start-workers.sh

访问 http://192.168.1.100:8080 查看Web UI，确认所有worker状态为ALIVE。

/opt/spark/bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://192.168.1.100:7077 \
  /opt/spark/examples/jars/spark-examples_2.12-3.4.0.jar \
  100

2.3 GraphX库引入

GraphX是Spark生态中原生的图计算库，内置于Spark核心发行包中，无需额外引入JAR包。只需在构建工具（如sbt或Maven）中声明Spark依赖即可。

2.3.1 Maven依赖配置

在pom.xml中添加以下依赖：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-graphx_2.12</artifactId>
    <version>3.4.0</version>
    <scope>provided</scope>
</dependency>

2.3.2 GraphX核心概念速览

GraphX将图表示为 Graph[VD, ED]，其中VD是顶点属性类型，ED是边属性类型。其底层基于RDD实现，天然支持分布式容错。

2.4 Neo4j与Spark集成

Neo4j作为业界领先的图数据库，存储着海量的反欺诈图谱数据。通过Neo4j Spark Connector，我们可以将Neo4j中的图谱数据直接加载为Spark DataFrame或GraphX图，实现“存储计算分离”的架构。

2.4.1 Neo4j安装与启动

在master节点上安装Neo4j Community版：

wget https://neo4j.com/artifact.php?name=neo4j-community-5.12.0-unix.tar.gz
tar -xzf neo4j-community-5.12.0-unix.tar.gz -C /opt/
/opt/neo4j-community-5.12.0/bin/neo4j start

修改/opt/neo4j-community-5.12.0/conf/neo4j.conf，开启Bolt连接：

server.bolt.enabled=true
server.bolt.listen_address=0.0.0.0:7687
server.bolt.advertised_address=192.168.1.100:7687

2.4.2 Spark Connector配置

在Spark作业中引入Neo4j Connector，有两种方式：

• 方式一（推荐）： 使用--packages参数自动下载。
• 方式二： 手动下载JAR并放入$SPARK_HOME/jars目录。

# 方式一：提交作业时自动解析依赖
spark-submit \
  --packages org.neo4j:neo4j-connector-apache-spark_2.12:5.1.0 \
  --master spark://192.168.1.100:7077 \
  --conf spark.neo4j.bolt.url=bolt://192.168.1.100:7687 \
  --conf spark.neo4j.bolt.user=neo4j \
  --conf spark.neo4j.bolt.password=password \
  your-fraud-detection-app.jar

2.4.3 数据读写示例

以下Scala代码演示了如何从Neo4j读取图谱数据并转换为GraphX图：

import org.apache.spark.graphx._
import org.apache.spark.sql.SparkSession
import org.neo4j.spark._

val spark = SparkSession.builder()
  .appName("Neo4jToGraphX")
  .config("spark.neo4j.bolt.url", "bolt://192.168.1.100:7687")
  .config("spark.neo4j.bolt.user", "neo4j")
  .config("spark.neo4j.bolt.password", "password")
  .getOrCreate()

// 读取Neo4j中的顶点和边
val vertexDF = spark.read.format("org.neo4j.spark.DataSource")
  .option("query", "MATCH (n:Device) RETURN id(n) AS id, n.device_id AS deviceId")
  .load()

val edgeDF = spark.read.format("org.neo4j.spark.DataSource")
  .option("query", "MATCH (u:User)-[r:LINKED_TO]->(d:Device) RETURN id(u) AS src, id(d) AS dst, r.timestamp AS ts")
  .load()

// 转换为GraphX图
val vertices: RDD[(VertexId, String)] = vertexDF.rdd.map(row => (row.getLong(0), row.getString(1)))
val edges: RDD[Edge[Long]] = edgeDF.rdd.map(row => Edge(row.getLong(0), row.getLong(1), row.getLong(2)))
val graph: Graph[String, Long] = Graph(vertices, edges)

2.5 开发环境验证

为了确认整个环境（Spark集群 + GraphX + Neo4j集成）工作正常，我们设计一个端到端的验证任务：从Neo4j加载一个简单的“用户-设备”二部图，在Spark中计算每个用户的设备度数，并将结果写回Neo4j。

2.5.1 验证流程图

2.5.2 验证步骤与代码

1. 准备Neo4j测试数据： 在Neo4j Browser中执行以下Cypher语句。

CREATE (u1:User {user_id: 'U001', name: 'Alice'})
CREATE (u2:User {user_id: 'U002', name: 'Bob'})
CREATE (d1:Device {device_id: 'D001', type: 'mobile'})
CREATE (d2:Device {device_id: 'D002', type: 'pc'})
CREATE (u1)-[:LINKED_TO {timestamp: 1700000000}]->(d1)
CREATE (u1)-[:LINKED_TO {timestamp: 1700000001}]->(d2)
CREATE (u2)-[:LINKED_TO {timestamp: 1700000002}]->(d1)

2. 提交验证作业： 使用spark-submit提交以下Scala代码。

import org.apache.spark.graphx._
import org.apache.spark.sql.SparkSession
import org.neo4j.spark._

object EnvValidation {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("EnvValidation")
      .config("spark.neo4j.bolt.url", "bolt://192.168.1.100:7687")
      .config("spark.neo4j.bolt.user", "neo4j")
      .config("spark.neo4j.bolt.password", "password")
      .getOrCreate()

    import spark.implicits._

    // 1. 读取顶点
    val vertexDF = spark.read.format("org.neo4j.spark.DataSource")
      .option("query", "MATCH (n) RETURN id(n) AS id, n.user_id AS userId, n.device_id AS deviceId")
      .load()
      .na.fill("")

    val vertices = vertexDF.rdd.map { row =>
      val id = row.getLong(0)
      val label = if (row.getString(1) != "") row.getString(1) else row.getString(2)
      (id, label)
    }

    // 2. 读取边
    val edgeDF = spark.read.format("org.neo4j.spark.DataSource")
      .option("query", "MATCH ()-[r:LINKED_TO]->() RETURN id(r) AS id, id(startNode(r)) AS src, id(endNode(r)) AS dst")
      .load()

    val edges = edgeDF.rdd.map(row => Edge(row.getLong(1), row.getLong(2), row.getLong(0)))

    // 3. 构建图并计算度数
    val graph = Graph(vertices, edges)
    val degrees = graph.degrees

    // 4. 将度数结果写回Neo4j
    val resultDF = degrees.toDF("vertexId", "degree")
    resultDF.write.format("org.neo4j.spark.DataSource")
      .option("save.mode", "Overwrite")
      .option("labels", "DegreeResult")
      .option("node.keys", "vertexId")
      .mode("overwrite")
      .save()

    // 5. 打印结果到控制台
    resultDF.show()

    spark.stop()
  }
}

3. 验证结果： 在Neo4j Browser中执行 MATCH (n:DegreeResult) RETURN n，应看到三个节点的度数：U001度数为2，U002度数为1，D001度数为2，D002度数为1。

2.6 常见问题与排错指南