Apache Spark Sedona：详细部署文档和应用探索指南（优化版）

文档版本： V2.0
编写人员： 郭宁泰
文档日期： 2025年12月9日
适用范围： 单机部署、集群化部署和应用探索

📋 目录

• 1. 概述
  • 1.1 Apache Sedona简介
  • 1.2 为什么需要Sedona
  • 1.3 文档目标
• 2. 版本管理
• 3. 部署环境分析
  • 3.1 部署方式对比
  • 3.2 版本适配推荐
• 4. 单机部署指南
  • 4.1 前置条件检查
  • 4.2 安装Java JDK
  • 4.3 安装Apache Spark
  • 4.4 配置Hadoop Winutils
  • 4.5 安装Python
  • 4.6 安装PySpark和Sedona
  • 4.7 测试验证
• 5. 集群化部署指南（生产环境）
  • 5.1 集群化部署架构
  • 5.2 集群化部署关键注意事项
  • 5.3 单机与集群的核心区别
  • 5.4 集群化部署步骤
  • 5.5 集群配置最佳实践
• 6. Python空间数据分析技术栈交互原理
  • 6.1 整体架构
  • 6.2 技术模块交互流程
  • 6.3 底层原理详解
  • 6.4 实例分析
• 7. 应用示例
  • 7.1 基础空间查询
  • 7.2 空间连接操作
  • 7.3 性能优化技巧
• 8. 常见问题与解决方案
• 9. 附录

1. 概述

1.1 Apache Sedona简介

Apache Sedona（原名GeoSpark）是一个用于处理大规模空间数据的集群计算系统，它扩展了Apache Spark的功能，提供了一套完整的空间数据类型、空间索引、空间分区和空间查询处理能力。

核心特性：

• 🌍 支持多种空间数据格式（GeoJSON、WKT、WKB、Shapefile等）
• 📊 分布式空间索引（R-Tree、Quad-Tree）
• 🔍 丰富的空间查询操作（范围查询、KNN、空间连接）
• ⚡ 高性能的分布式空间计算
• 🐍 多语言支持（Scala、Java、Python、R）

1.2 为什么需要Sedona

在处理大规模地理空间数据时，传统的GIS工具（如PostGIS、ArcGIS）面临以下挑战：

• 数据规模限制：单机处理能力有限，难以处理TB级以上数据
• 计算效率低：复杂的空间分析需要数小时甚至数天
• 扩展性差：难以利用分布式计算资源

Sedona通过与Spark集成，解决了这些问题：

• ✅ 分布式处理，可水平扩展到数百个节点
• ✅ 内存计算，大幅提升处理速度
• ✅ 与大数据生态无缝集成（HDFS、Hive、Kafka等）

1.3 文档目标

本文档旨在提供：

1. 详细的单机部署指导（适合开发和测试）
2. 完整的集群化部署方案（适合生产环境）
3. 技术栈交互原理深度解析
4. 实用的应用示例和最佳实践

2. 版本管理

3. 部署环境分析

3.1 部署方式对比

3.2 版本适配推荐

4. 单机部署指南

4.1 前置条件检查

系统要求：

• 操作系统：Windows 10/11（专业版/企业版）、Linux、macOS
• 内存：≥ 8GB（建议16GB）
• 磁盘空间：≥ 10GB
• 网络：可访问外网（用于下载依赖）

检查虚拟化支持（Docker部署需要）：

# Windows系统检查
systeminfo | findstr /C:"Hyper-V"

4.2 安装Java JDK

版本选择： Java JDK 11（推荐）

下载地址：

https://github.com/adoptium/temurin11-binaries/releases/download/jdk-11.0.29+7/OpenJDK11U-jdk_x64_windows_hotspot_11.0.29_7.msi

安装步骤：

1. 检查现有Java环境

java -version

2. 卸载旧版本（如果存在）

   • Windows：控制面板 → 程序和功能
   • Linux：sudo apt remove openjdk-*

3. 安装JDK 11

   • 双击MSI安装包
   • 选择默认路径：C:\Program Files\Eclipse Adoptium\jdk-11

4. 配置环境变量（PowerShell管理员）

# 设置JAVA_HOME
[System.Environment]::SetEnvironmentVariable("JAVA_HOME", "C:\Program Files\Eclipse Adoptium\jdk-11", "Machine")

# 添加到PATH
$oldPath = [System.Environment]::GetEnvironmentVariable("Path", "Machine")
$newPath = $oldPath + ";C:\Program Files\Eclipse Adoptium\jdk-11\bin"
[System.Environment]::SetEnvironmentVariable("Path", $newPath, "Machine")

5. 验证安装

java -version
javac -version
echo %JAVA_HOME%

预期输出：

openjdk version "11.0.29" 2024-10-15
OpenJDK Runtime Environment Temurin-11.0.29+7 (build 11.0.29+7)

4.3 安装Apache Spark

版本选择： Spark 3.4.0

下载地址：

https://archive.apache.org/dist/spark/spark-3.4.0/spark-3.4.0-bin-hadoop3.tgz

安装步骤：

1. 解压Spark包

# 解压到指定目录，例如 D:\sp_project\spark-3.4.0

2. 配置环境变量（PowerShell管理员）

# 设置SPARK_HOME
[System.Environment]::SetEnvironmentVariable("SPARK_HOME", "D:\sp_project\spark-3.4.0", "Machine")

# 添加到PATH
$oldPath = [System.Environment]::GetEnvironmentVariable("Path", "Machine")
$newPath = $oldPath + ";D:\sp_project\spark-3.4.0\bin"
[System.Environment]::SetEnvironmentVariable("Path", $newPath, "Machine")

3. 验证安装

spark-submit --version

4.4 配置Hadoop Winutils

版本选择： Hadoop 3.0

下载地址：

https://github.com/steveloughran/winutils/blob/master/hadoop-3.0.0/bin/winutils.exe

配置步骤：

1. 创建Hadoop目录

# 例如：D:\sp_project\hadoop\bin
# 将winutils.exe放入bin目录

2. 配置环境变量

[System.Environment]::SetEnvironmentVariable("HADOOP_HOME", "D:\sp_project\hadoop", "Machine")

# 添加到PATH
$oldPath = [System.Environment]::GetEnvironmentVariable("Path", "Machine")
$newPath = $oldPath + ";D:\sp_project\hadoop\bin"
[System.Environment]::SetEnvironmentVariable("Path", $newPath, "Machine")

4.5 安装Python

版本选择： Python 3.11

下载地址：

https://www.python.org/ftp/python/3.11.6/python-3.11.6-amd64.exe

安装步骤：

1. 检查现有Python环境

python --version

2. 安装Python 3.11

   • 勾选 "Add Python to PATH"
   • 选择 "Customize installation"
   • 推荐安装路径：D:\sp_project\python3.11

3. 配置环境变量

[System.Environment]::SetEnvironmentVariable("PYSPARK_PYTHON", "python", "Machine")

4. 验证安装

python --version
pip --version

4.6 安装PySpark和Sedona

安装步骤：

1. 升级pip

python -m pip install --upgrade pip

2. 安装Apache Sedona

pip install apache-sedona==1.8.0

3. 安装依赖库

pip install pandas geopandas matplotlib jupyter pyspark==3.4.0

4. 下载Sedona JAR包（可选，用于Spark集成）

# 下载地址
https://repo1.maven.org/maven2/org/apache/sedona/sedona-spark-3.4_2.12/1.8.0/

4.7 测试验证

创建测试脚本：

# test_sedona.py
from pyspark.sql import SparkSession
from sedona.spark import SedonaContext

# 创建Spark会话
spark = SparkSession.builder \
    .appName("Sedona Test") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrator", "org.apache.sedona.core.serde.SedonaKryoRegistrator") \
    .getOrCreate()

# 初始化Sedona上下文
sedona = SedonaContext.create(spark)

# 创建测试数据
test_data = [
    ("POINT (1 1)",),
    ("POINT (2 2)",),
    ("POINT (3 3)",)
]

df = sedona.createDataFrame(test_data, ["geometry"])
df.show()

print("✅ Sedona安装成功！")
spark.stop()

运行测试：

python test_sedona.py

5. 集群化部署指南（生产环境）

⚠️ 重要说明：系统选择

• Windows系统：适合单机开发和测试（见第4章）
• Linux系统：适合集群生产环境（本章内容）

Windows集群部署的限制：

• Spark Standalone集群在Windows上支持有限
• 建议使用以下替代方案：
  1. Docker Desktop - 在Windows上运行Linux容器集群
  2. WSL2 + Linux集群 - 在Windows子系统中部署
  3. 云平台 - AWS EMR、Azure HDInsight（推荐生产环境）
  4. 虚拟机方案 - VMware/VirtualBox + Linux集群

本章主要介绍Linux集群部署。如需Windows环境方案，请参考5.6 Windows环境集群方案。

5.1 集群化部署架构

┌─────────────────────────────────────────────────────────────────┐
│                        集群化部署架构                              │
└─────────────────────────────────────────────────────────────────┘

                    ┌──────────────────┐
                    │   客户端应用      │
                    │  (Python/Scala)  │
                    └────────┬─────────┘
                             │
                    ┌────────▼─────────┐
                    │   Spark Driver    │
                    │  (调度与协调)      │
                    └────────┬─────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
    ┌─────────▼────┐  ┌─────▼──────┐  ┌───▼────────┐
    │ Worker Node 1│  │Worker Node 2│  │Worker Node N│
    │              │  │             │  │             │
    │ ┌──────────┐ │  │ ┌──────────┐│  │ ┌──────────┐│
    │ │Executor 1│ │  │ │Executor 2││  │ │Executor N││
    │ │+ Sedona  │ │  │ │+ Sedona  ││  │ │+ Sedona  ││
    │ └──────────┘ │  │ └──────────┘│  │ └──────────┘│
    └──────┬───────┘  └──────┬──────┘  └──────┬──────┘
           │                 │                 │
           └─────────────────┼─────────────────┘
                             │
                    ┌────────▼─────────┐
                    │   HDFS/S3/OSS     │
                    │  (分布式存储)      │
                    └──────────────────┘

5.2 集群化部署关键注意事项

5.2.1 资源管理与配置

1. 内存配置

# spark-defaults.conf

# Driver内存（建议：集群总内存的5-10%）
spark.driver.memory              4g
spark.driver.memoryOverhead      1g

# Executor内存（关键配置）
spark.executor.memory            8g
spark.executor.memoryOverhead    2g

# Executor数量和核心数
spark.executor.instances         10
spark.executor.cores             4

# 动态资源分配
spark.dynamicAllocation.enabled  true
spark.dynamicAllocation.minExecutors  2
spark.dynamicAllocation.maxExecutors  20

内存计算公式：

单节点可用内存 = 节点总内存 × 0.75（保留25%给系统）
每个Executor内存 = (单节点可用内存) / (每节点Executor数)

2. 存储配置

# 数据本地化
spark.locality.wait              3s

# Shuffle配置
spark.sql.shuffle.partitions     200
spark.shuffle.service.enabled    true

# 序列化配置（Sedona必需）
spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.kryo.registrator          org.apache.sedona.core.serde.SedonaKryoRegistrator

5.2.2 网络与通信

1. 端口配置

2. 网络优化

# 网络超时配置
spark.network.timeout            600s
spark.rpc.askTimeout            300s

# 数据传输优化
spark.reducer.maxSizeInFlight    96m
spark.shuffle.io.maxRetries      5

5.2.3 数据分区与索引

1. Sedona空间分区策略

# 空间分区配置
from sedona.spark import SedonaContext
from sedona.core.enums import GridType, IndexType

# 设置分区数（建议：Executor数量 × 2-4倍）
spatial_partitions = 40

# 使用KDB-Tree分区（推荐用于空间数据）
spatial_rdd.spatialPartitioning(GridType.KDBTREE, spatial_partitions)

# 构建空间索引
spatial_rdd.buildIndex(IndexType.RTREE, True)

2. 分区策略选择

5.2.4 高可用性配置

1. Spark集群高可用

# 使用ZooKeeper实现Master高可用
spark.deploy.recoveryMode        ZOOKEEPER
spark.deploy.zookeeper.url       zk1:2181,zk2:2181,zk3:2181
spark.deploy.zookeeper.dir       /spark

2. HDFS高可用

<!-- hdfs-site.xml -->
<configuration>
    <property>
        <name>dfs.nameservices</name>
        <value>mycluster</value>
    </property>
    <property>
        <name>dfs.ha.namenodes.mycluster</name>
        <value>nn1,nn2</value>
    </property>
</configuration>

5.2.5 安全配置

1. 认证与授权

# Kerberos认证
spark.authenticate               true
spark.authenticate.secret        your-secret-key

# SSL加密
spark.ssl.enabled                true
spark.ssl.keyStore              /path/to/keystore
spark.ssl.keyStorePassword      password

2. 网络隔离

• 使用VPC隔离集群网络
• 配置安全组规则
• 启用防火墙策略

5.2.6 监控与日志

1. 监控配置

# 启用Metrics
spark.metrics.conf              /path/to/metrics.properties

# 事件日志
spark.eventLog.enabled          true
spark.eventLog.dir              hdfs://namenode:9000/spark-logs

# History Server
spark.history.fs.logDirectory   hdfs://namenode:9000/spark-logs

2. 推荐监控工具

• Spark History Server（内置）
• Ganglia（集群监控）
• Prometheus + Grafana（推荐）
• ELK Stack（日志分析）

5.3 单机与集群的核心区别

5.3.1 架构层面

5.3.2 数据处理差异

单机模式：

# 本地模式启动
spark = SparkSession.builder \
    .master("local[*]") \
    .appName("Local Test") \
    .getOrCreate()

# 读取本地文件
df = spark.read.csv("file:///path/to/local/data.csv")

集群模式：

# 集群模式启动
spark = SparkSession.builder \
    .master("spark://master:7077") \
    .appName("Cluster App") \
    .config("spark.executor.instances", "10") \
    .getOrCreate()

# 读取HDFS文件
df = spark.read.csv("hdfs://namenode:9000/data/data.csv")

5.3.3 性能差异

数据规模对比：

5.3.4 开发与运维差异

开发流程：

单机开发 → 本地测试 → 集群测试 → 生产部署
   ↓          ↓          ↓          ↓
 快速迭代   功能验证   性能调优   稳定运行

运维复杂度：

5.4 集群化部署步骤

步骤1：准备集群节点

节点规划：

Master节点（1-2台）：
- CPU: 8核+
- 内存: 16GB+
- 磁盘: 500GB SSD

Worker节点（N台）：
- CPU: 16核+
- 内存: 64GB+
- 磁盘: 2TB+ HDD（数据存储）

步骤2：安装基础环境（所有节点）

# 1. 安装Java
sudo apt update
sudo apt install openjdk-11-jdk -y

# 2. 配置SSH免密登录
ssh-keygen -t rsa
ssh-copy-id user@worker-node

# 3. 同步时间
sudo apt install ntp -y
sudo systemctl start ntp

# 4. 配置hosts
sudo vim /etc/hosts
# 添加所有节点的IP和主机名

步骤3：安装Hadoop HDFS

# 下载Hadoop
wget https://archive.apache.org/dist/hadoop/common/hadoop-3.3.4/hadoop-3.3.4.tar.gz
tar -xzf hadoop-3.3.4.tar.gz
sudo mv hadoop-3.3.4 /opt/hadoop

# 配置环境变量
echo 'export HADOOP_HOME=/opt/hadoop' >> ~/.bashrc
echo 'export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin' >> ~/.bashrc
source ~/.bashrc

# 配置core-site.xml
vim $HADOOP_HOME/etc/hadoop/core-site.xml
# 添加HDFS配置

# 配置hdfs-site.xml
vim $HADOOP_HOME/etc/hadoop/hdfs-site.xml
# 添加副本和存储配置

# 格式化NameNode（仅首次）
hdfs namenode -format

# 启动HDFS
start-dfs.sh

步骤4：安装Spark集群

# 下载Spark
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
sudo mv spark-3.4.0-bin-hadoop3 /opt/spark

# 配置环境变量
echo 'export SPARK_HOME=/opt/spark' >> ~/.bashrc
echo 'export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin' >> ~/.bashrc
source ~/.bashrc

# 配置spark-env.sh
cd $SPARK_HOME/conf
cp spark-env.sh.template spark-env.sh
vim spark-env.sh

# 添加以下配置
export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64
export HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
export SPARK_MASTER_HOST=master-node
export SPARK_MASTER_PORT=7077

# 配置workers文件
vim workers
# 添加所有worker节点的主机名
worker1
worker2
worker3

# 将配置同步到所有节点
scp -r $SPARK_HOME user@worker1:/opt/
scp -r $SPARK_HOME user@worker2:/opt/
scp -r $SPARK_HOME user@worker3:/opt/

步骤5：安装Sedona到集群

# 在所有节点安装Python和依赖
sudo apt install python3-pip -y
pip3 install apache-sedona==1.8.0 pyspark==3.4.0

# 下载Sedona JAR包
cd $SPARK_HOME/jars
wget https://repo1.maven.org/maven2/org/apache/sedona/sedona-spark-3.4_2.12/1.8.0/sedona-spark-3.4_2.12-1.8.0.jar
wget https://repo1.maven.org/maven2/org/locationtech/jts/jts-core/1.19.0/jts-core-1.19.0.jar

# 同步到所有节点
for node in worker1 worker2 worker3; do
    scp *.jar user@$node:$SPARK_HOME/jars/
done

步骤6：启动集群

# 在Master节点启动
$SPARK_HOME/sbin/start-master.sh

# 启动所有Worker节点
$SPARK_HOME/sbin/start-workers.sh

# 或在每个Worker节点单独启动
$SPARK_HOME/sbin/start-worker.sh spark://master-node:7077

# 验证集群状态
# 访问 http://master-node:8080

5.5 集群配置最佳实践

5.5.1 资源配置建议

小型集群（3-5节点）：

spark.executor.instances         6
spark.executor.cores             4
spark.executor.memory            8g
spark.driver.memory              4g
spark.sql.shuffle.partitions     100

中型集群（10-20节点）：

spark.executor.instances         20
spark.executor.cores             4
spark.executor.memory            16g
spark.driver.memory              8g
spark.sql.shuffle.partitions     200

大型集群（50+节点）：

spark.executor.instances         100
spark.executor.cores             5
spark.executor.memory            20g
spark.driver.memory              16g
spark.sql.shuffle.partitions     500

5.5.2 Sedona特定优化

# 空间数据优化配置
spark.conf.set("sedona.join.numpartition", 200)
spark.conf.set("sedona.join.gridtype", "kdbtree")
spark.conf.set("sedona.join.indexbuildside", "left")

# 内存优化
spark.conf.set("spark.memory.fraction", 0.8)
spark.conf.set("spark.memory.storageFraction", 0.3)

5.5.3 监控指标

关键监控指标：

• CPU使用率（建议 < 80%）
• 内存使用率（建议 < 85%）
• 磁盘I/O（关注读写延迟）
• 网络带宽（shuffle时关注）
• GC时间（建议 < 10%的总时间）
• Task失败率（建议 < 1%）

5.6 Windows环境集群方案

由于Spark在Windows上原生集群支持有限，这里提供几种在Windows环境中实现类似集群效果的方案：

5.6.1 方案1：Docker Desktop（推荐）

适用场景： 开发测试、小规模数据处理

前置条件：

• Windows 10/11 专业版或企业版
• 开启Hyper-V或WSL2
• 安装Docker Desktop

部署步骤：

1. 安装Docker Desktop

# 下载并安装Docker Desktop for Windows
# https://www.docker.com/products/docker-desktop

# 启动Docker Desktop
# 设置 → Resources → WSL Integration → 启用

2. 创建docker-compose.yml

# docker-compose.yml
version: '3'
services:
  spark-master:
    image: bitnami/spark:3.4.0
    environment:
      - SPARK_MODE=master
      - SPARK_MASTER_HOST=spark-master
    ports:
      - "8080:8080"
      - "7077:7077"
    volumes:
      - ./data:/data
      - ./jars:/opt/bitnami/spark/jars/custom
    networks:
      - spark-network

  spark-worker-1:
    image: bitnami/spark:3.4.0
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=4G
      - SPARK_WORKER_CORES=2
    depends_on:
      - spark-master
    networks:
      - spark-network

  spark-worker-2:
    image: bitnami/spark:3.4.0
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=4G
      - SPARK_WORKER_CORES=2
    depends_on:
      - spark-master
    networks:
      - spark-network

networks:
  spark-network:
    driver: bridge

3. 启动集群

# 在docker-compose.yml所在目录
docker-compose up -d

# 查看状态
docker-compose ps

# 访问Web UI: http://localhost:8080

4. 安装Sedona到容器

# 下载Sedona JAR包到本地jars目录
mkdir jars
cd jars

# 使用PowerShell下载
Invoke-WebRequest -Uri "https://repo1.maven.org/maven2/org/apache/sedona/sedona-spark-3.4_2.12/1.8.0/sedona-spark-3.4_2.12-1.8.0.jar" -OutFile "sedona-spark-3.4_2.12-1.8.0.jar"
Invoke-WebRequest -Uri "https://repo1.maven.org/maven2/org/locationtech/jts/jts-core/1.19.0/jts-core-1.19.0.jar" -OutFile "jts-core-1.19.0.jar"

# 重启容器加载JAR包
docker-compose restart

5. 连接到集群（从Windows主机）

# 在Windows主机上运行
from pyspark.sql import SparkSession
from sedona.spark import SedonaContext

spark = SparkSession.builder \
    .appName("Sedona Docker Cluster") \
    .master("spark://localhost:7077") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrator", "org.apache.sedona.core.serde.SedonaKryoRegistrator") \
    .getOrCreate()

sedona = SedonaContext.create(spark)
print("✅ 连接到Docker集群成功！")

优点：

• ✅ 在Windows上快速搭建集群环境
• ✅ 环境隔离，不影响主机系统
• ✅ 易于扩展Worker数量

限制：

• ⚠️ 性能受Docker虚拟化影响
• ⚠️ 仅适合中小规模数据（< 100GB）

5.6.2 方案2：WSL2 + Linux集群

适用场景： 性能要求较高的开发测试

部署步骤：

1. 启用WSL2

# 以管理员身份运行PowerShell
wsl --install

# 安装Ubuntu
wsl --install -d Ubuntu-22.04

# 设置默认版本为WSL2
wsl --set-default-version 2

2. 在WSL2中部署集群

# 进入WSL2
wsl

# 按照第5.4节的Linux部署步骤进行
# 在WSL2中完全按照Linux方式部署Spark集群

3. 从Windows访问WSL2集群

# Windows主机上的Python脚本
spark = SparkSession.builder \
    .master("spark://localhost:7077") \  # WSL2端口自动转发
    .getOrCreate()

优点：

• ✅ 接近原生Linux性能
• ✅ 完整的Linux开发环境
• ✅ 可直接使用Linux部署文档

注意事项：

• 需要在WSL2中配置网络端口转发
• WSL2重启后IP可能变化

5.6.3 方案3：云平台（生产推荐）

适用场景： 生产环境、大规模数据处理

AWS EMR（推荐）

1. 创建EMR集群

# 使用AWS CLI（在Windows PowerShell中）
aws emr create-cluster `
  --name "Sedona-Cluster" `
  --release-label emr-6.10.0 `
  --applications Name=Spark `
  --ec2-attributes KeyName=mykey `
  --instance-type m5.xlarge `
  --instance-count 3 `
  --use-default-roles

2. 安装Sedona

# SSH连接到EMR主节点
ssh -i mykey.pem hadoop@<master-public-dns>

# 安装Sedona
sudo pip3 install apache-sedona==1.8.0

# 下载JAR包到Spark
cd /usr/lib/spark/jars
sudo wget https://repo1.maven.org/maven2/org/apache/sedona/sedona-spark-3.4_2.12/1.8.0/sedona-spark-3.4_2.12-1.8.0.jar

3. 从Windows连接到EMR

# Windows主机上连接
spark = SparkSession.builder \
    .master("spark://<emr-master-dns>:7077") \
    .getOrCreate()

Azure HDInsight

# 使用Azure CLI创建集群
az hdinsight create `
  --name sedona-cluster `
  --resource-group myResourceGroup `
  --type spark `
  --component-version Spark=3.1 `
  --size Standard_D13_V2 `
  --worker-node-count 3

优点：

• ✅ 完全托管，无需维护基础设施
• ✅ 弹性伸缩，按需付费
• ✅ 企业级安全和监控

成本：

• 按使用时间和资源计费
• 建议先使用免费试用额度

5.6.4 方案4：虚拟机集群

适用场景： 学习、测试完整集群环境

使用VirtualBox/VMware：

1. 创建Linux虚拟机

Master节点: 2核8GB, Ubuntu 22.04
Worker1: 2核4GB, Ubuntu 22.04  
Worker2: 2核4GB, Ubuntu 22.04

2. 网络配置

• 使用桥接网络或Host-Only网络
• 配置固定IP地址
• 确保各虚拟机可互相访问

3. 按Linux方式部署

• 按照5.4节步骤完整部署

优点：

• ✅ 完整模拟真实集群环境
• ✅ 适合学习和测试

限制：

• ⚠️ 占用大量硬件资源
• ⚠️ 性能较低

5.6.5 方案对比与选择建议

推荐选择：

开发阶段 → Docker Desktop或WSL2
  ↓
测试阶段 → WSL2集群或小规模云集群
  ↓
生产部署 → 云平台（AWS EMR/Azure HDInsight）

5.6.6 Windows特定注意事项

1. 路径分隔符

# Windows路径需要转义或使用原始字符串
data_path = r"D:\data\spatial_data.csv"  # 推荐
# 或
data_path = "D:/data/spatial_data.csv"   # 使用正斜杠

2. 环境变量设置

# 使用PowerShell设置环境变量
[System.Environment]::SetEnvironmentVariable("SPARK_HOME", "D:\spark", "Machine")

3. 文件权限

# Windows不需要chmod，但可能需要以管理员身份运行
# 右键点击PowerShell → "以管理员身份运行"

4. 防火墙配置

# 添加Spark端口到防火墙例外
New-NetFirewallRule -DisplayName "Spark Master" -Direction Inbound -LocalPort 7077 -Protocol TCP -Action Allow
New-NetFirewallRule -DisplayName "Spark Web UI" -Direction Inbound -LocalPort 8080 -Protocol TCP -Action Allow

5. 本地文件URI格式

# Windows本地文件
df = spark.read.csv("file:///D:/data/points.csv")  # 注意三个斜杠

6. Python空间数据分析技术栈交互原理

6.1 整体架构

┌────────────────────────────────────────────────────────────────────┐
│           Python空间数据分析完整技术栈架构图                           │
└────────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│                    应用层 (Application Layer)                      │
├──────────────────────────────────────────────────────────────────┤
│  Python脚本/Jupyter Notebook                                      │
│  ├─ 数据加载: pd.read_csv(), gpd.read_file()                      │
│  ├─ 空间分析: spatial join, buffer, intersection                 │
│  └─ 结果可视化: matplotlib, folium                                │
└───────────────────────┬──────────────────────────────────────────┘
                        │ Python API调用
                        ▼
┌──────────────────────────────────────────────────────────────────┐
│                 PySpark层 (PySpark Layer)                         │
├──────────────────────────────────────────────────────────────────┤
│  SparkSession / SedonaContext                                     │
│  ├─ DataFrame API: df.filter(), df.join()                        │
│  ├─ Sedona SQL: ST_Point(), ST_Distance(), ST_Contains()         │
│  └─ RDD API: spatialRDD.spatialPartitioning()                    │
└───────────────────────┬──────────────────────────────────────────┘
                        │ Py4J网关
                        ▼
┌──────────────────────────────────────────────────────────────────┐
│              Spark Driver (JVM进程 - 调度器)                       │
├──────────────────────────────────────────────────────────────────┤
│  Spark Core                                                       │
│  ├─ DAG调度器: 将逻辑计划转为物理执行计划                           │
│  ├─ Task调度器: 分配任务到Executor                                 │
│  └─ 集群管理器接口: 与YARN/Mesos/K8s通信                           │
│                                                                   │
│  Sedona Core (Scala/Java)                                        │
│  ├─ 空间数据类型: Geometry, Point, Polygon                         │
│  ├─ 空间索引: R-Tree, Quad-Tree构建                               │
│  ├─ 空间分区: KDB-Tree, Equal Partition                          │
│  └─ 空间查询优化器                                                 │
└───────────────────────┬──────────────────────────────────────────┘
                        │ RPC调用 (Netty)
                        ▼
┌──────────────────────────────────────────────────────────────────┐
│           Spark Executors (多个JVM进程 - 计算节点)                 │
├──────────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │
│  │ Executor 1  │  │ Executor 2  │  │ Executor N  │              │
│  ├─────────────┤  ├─────────────┤  ├─────────────┤              │
│  │Task执行引擎 │  │Task执行引擎 │  │Task执行引擎 │              │
│  │+ Sedona函数 │  │+ Sedona函数 │  │+ Sedona函数 │              │
│  │            │  │            │  │            │              │
│  │空间索引缓存 │  │空间索引缓存 │  │空间索引缓存 │              │
│  │R-Tree/QTree│  │R-Tree/QTree│  │R-Tree/QTree│              │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘              │
│         │ 读写            │ 读写            │ 读写                │
└─────────┼─────────────────┼─────────────────┼────────────────────┘
          │                 │                 │
          ▼                 ▼                 ▼
┌──────────────────────────────────────────────────────────────────┐
│              存储层 (Storage Layer)                                │
├──────────────────────────────────────────────────────────────────┤
│  Hadoop HDFS (分布式文件系统)                                       │
│  ├─ NameNode: 元数据管理                                          │
│  │   └─ 文件目录树、数据块位置映射                                  │
│  ├─ DataNode: 实际数据存储                                        │
│  │   ├─ DataNode 1: Block 1, 3, 5...                            │
│  │   ├─ DataNode 2: Block 2, 4, 6...                            │
│  │   └─ DataNode 3: Block 1, 2, 7... (副本)                     │
│  └─ 数据副本机制（默认3副本）                                       │
│                                                                   │
│  或其他存储:                                                       │
│  ├─ AWS S3 / 阿里云OSS                                            │
│  ├─ 本地文件系统 (file://)                                        │
│  └─ Hive / HBase                                                 │
└──────────────────────────────────────────────────────────────────┘

6.2 技术模块交互流程

6.2.1 完整执行流程（13个步骤）

步骤详解：Python空间分析命令的完整生命周期

用户执行：python spatial_analysis.py

┌─────────────────────────────────────────────────────────────────┐
│ 阶段1: 初始化与连接建立                                            │
└─────────────────────────────────────────────────────────────────┘

[步骤1] Python进程启动
├─ 加载PySpark库
├─ 启动Py4J网关 (默认端口25333)
└─ 建立Python到JVM的通信桥梁

[步骤2] Spark Driver进程启动 (JVM)
├─ 初始化SparkContext
├─ 加载Sedona库 (sedona-spark-*.jar)
├─ 注册Sedona空间函数到Spark SQL
└─ 连接到集群管理器 (YARN/Standalone/K8s)

[步骤3] 建立与Hadoop的连接
├─ 读取Hadoop配置 (core-site.xml, hdfs-site.xml)
├─ 连接HDFS NameNode (默认端口9000)
└─ 获取可用DataNode列表

┌─────────────────────────────────────────────────────────────────┐
│ 阶段2: 数据加载与解析                                              │
└─────────────────────────────────────────────────────────────────┘

[步骤4] Python代码读取数据
Python代码示例:
```python
# 读取HDFS上的空间数据
df = spark.read.format("csv") \
    .option("header", "true") \
    .load("hdfs://namenode:9000/data/points.csv")

[步骤5] Spark Driver解析读取请求
├─ 将DataFrame操作转换为逻辑计划 (Logical Plan)
├─ Catalyst优化器进行优化
│ ├─ 谓词下推 (Predicate Pushdown)
│ ├─ 列裁剪 (Column Pruning)
│ └─ 常量折叠 (Constant Folding)
└─ 生成物理执行计划 (Physical Plan)

[步骤6] HDFS数据读取
├─ Driver向NameNode查询文件元数据
│ └─ 获取数据块位置: Block1在DataNode1,3; Block2在DataNode2,4
├─ Driver创建读取任务 (ReadTasks)
└─ 分配任务到Executor (考虑数据本地性)

[步骤7] Executor读取数据块
├─ Executor1从本地DataNode读取Block1 (本地读取, 快速)
├─ Executor2从本地DataNode读取Block2
├─ Executor3从远程DataNode读取Block3 (网络传输, 较慢)
└─ 数据加载到Executor内存 (RDD/DataFrame分区)

┌─────────────────────────────────────────────────────────────────┐
│ 阶段3: 空间数据处理 (Sedona核心) │
└─────────────────────────────────────────────────────────────────┘

[步骤8] 空间数据转换
Python代码示例:

from sedona.sql import st_functions as STF

# 将经纬度转换为空间点对象
df = df.withColumn("geometry", 
                   STF.ST_Point(df.longitude, df.lat))

在Executor中的处理:
├─ Sedona解析WKT/WKB格式
├─ 创建JTS Geometry对象 (Java Topology Suite)
├─ 构建空间对象 (Point, Polygon, LineString)
└─ 存储在DataFrame的Geometry列中

[步骤9] 空间分区 (Spatial Partitioning)
Python代码示例:

from sedona.core.enums import GridType

# 使用KDB-Tree进行空间分区
spatial_rdd.spatialPartitioning(GridType.KDBTREE, 100)

处理流程:
├─ Driver采样数据 (sample ~1%)
├─ 构建KDB-Tree分区边界
│ ├─ 递归二分空间
│ ├─ 保证每个分区数据量均衡
│ └─ 生成100个空间分区
├─ 将分区边界广播到所有Executor
└─ Executor根据几何对象位置分配到对应分区

[步骤10] 空间索引构建
Python代码示例:

from sedona.core.enums import IndexType

# 为每个分区构建R-Tree索引
spatial_rdd.buildIndex(IndexType.RTREE, True)

在Executor中并行构建:
├─ Executor1: 为分区1-10构建R-Tree
│ ├─ 插入几何对象到R-Tree
│ ├─ 自动平衡树结构
│ └─ 缓存索引到内存
├─ Executor2: 为分区11-20构建R-Tree
└─ ExecutorN: 为分区91-100构建R-Tree

R-Tree索引结构:

       Root (MBR: 全局边界)
      /    \
   Node1   Node2
   /  \     /  \
 Leaf1 Leaf2 Leaf3 Leaf4
 (包含实际几何对象)

┌─────────────────────────────────────────────────────────────────┐
│ 阶段4: 空间查询与计算 │
└─────────────────────────────────────────────────────────────────┘

[步骤11] 空间连接 (Spatial Join) 示例
Python代码:

# 查找每个点所在的区域
result = point_df.join(
    polygon_df,
    STF.ST_Contains(polygon_df.geometry, point_df.geometry)
)

执行流程:
├─ Driver生成Join执行计划
│ ├─ 选择Join策略: Broadcast Join vs Shuffle Join
│ └─ 确定索引使用策略
│
├─ 如果数据量小: Broadcast Join
│ ├─ 将小表(polygon_df)广播到所有Executor
│ ├─ 每个Executor本地Join
│ └─ 无需Shuffle, 性能最优
│
├─ 如果数据量大: Shuffle Join
│ ├─ Phase 1: Partition Pruning (分区剪枝)
│ │ └─ 利用空间分区, 只Join可能相交的分区对
│ │
│ ├─ Phase 2: Shuffle操作
│ │ ├─ Executor1将分区1数据发送到目标Executor
│ │ ├─ Executor2将分区2数据发送到目标Executor
│ │ └─ 网络传输 (Netty, 可能数GB数据)
│ │
│ └─ Phase 3: 本地Join
│ ├─ 利用R-Tree索引加速
│ ├─ 对每个point查询R-Tree
│ └─ 精确几何测试: JTS算法
│ ├─ ST_Contains: 点在多边形内测试
│ ├─ ST_Intersects: 相交测试
│ └─ ST_Distance: 距离计算
│
└─ 生成Join结果DataFrame

[步骤12] 空间聚合计算
Python代码:

# 计算每个区域的点数量
result = result.groupBy("region_id") \
               .agg(count("*").alias("point_count"))

执行流程:
├─ Driver创建聚合任务
├─ 各Executor本地预聚合 (Map-side aggregation)
├─ Shuffle聚合数据到少数Executor
└─ 最终聚合计算

┌─────────────────────────────────────────────────────────────────┐
│ 阶段5: 结果收集与返回 │
└─────────────────────────────────────────────────────────────────┘

[步骤13] 结果返回到Python
Python代码:

# 收集结果到Driver
result_pd = result.toPandas()

# 或保存到HDFS
result.write.parquet("hdfs://namenode:9000/output/result")

执行流程:
├─ 如果.toPandas():
│ ├─ 各Executor将结果发送到Driver
│ ├─ Driver合并所有结果
│ ├─ 通过Py4J转换为Python对象
│ └─ 创建Pandas DataFrame
│
└─ 如果.write():
├─ 各Executor直接写入HDFS
├─ 每个Executor写入一个或多个文件
└─ 无需经过Driver (分布式写入)

### 6.3 底层原理详解

#### 6.3.1 Py4J通信机制

```python
# Python侧
from pyspark.sql import SparkSession

# 创建SparkSession时, 实际上:
# 1. Python启动JVM进程
# 2. 通过Py4J Gateway建立socket连接
# 3. Python对象映射到Java对象

spark = SparkSession.builder.getOrCreate()

# 当你调用df.show()时:
df.show()
# ↓
# Python通过Py4J发送RPC调用
# ↓
# JVM接收调用, 执行Java/Scala代码
# ↓
# 结果通过Py4J返回Python

Py4J调用示例：

Python进程                          JVM进程
    │                                  │
    │  createDataFrame([1,2,3])       │
    ├─────────────────────────────────>│
    │                                  │ 执行Java代码
    │                                  │ 创建DataFrame
    │                                  │
    │  <返回DataFrame引用(JavaObject)>  │
    │<─────────────────────────────────┤
    │                                  │
    │  df.show()                       │
    ├─────────────────────────────────>│
    │                                  │ 执行显示逻辑
    │                                  │
    │  <返回显示结果字符串>             │
    │<─────────────────────────────────┤

6.3.2 Sedona空间索引原理

R-Tree索引结构：

┌─────────────────────────────────────────────────────────────────┐
│                     R-Tree索引工作原理                            │
└─────────────────────────────────────────────────────────────────┘

空间数据分布:
    Y
    │  ┌───────────────────────────┐
 10 │  │ A: Building              │
    │  │  ●●●                      │
  8 │  │   ●● ┌────────┐          │
    │  │      │ B: Park│          │
  6 │  │      │  ●●●●  │  C: Lake │
    │  │      │  ●●●   │  ●●●●●●  │
  4 │  │      └────────┘  ●●●●●   │
    │  │                  ●●●●    │
  2 │  │ D: Road ●●●●●●●●●        │
    │  │                          │
  0 │  └───────────────────────────┘
    └─────────────────────────────── X
    0  2   4   6   8  10  12  14

R-Tree索引结构:
                Root MBR (0,0,15,10)
               /                    \
      MBR1 (0,0,8,10)           MBR2 (8,0,15,10)
       /        \                   |
  A(0,8,5,10) B(4,4,8,8)       C(10,4,15,8)
  Buildings    Park              Lake
  
  Leaf Node D(0,2,12,2) - Road

查询: "查找与矩形(3,3,7,7)相交的对象"
1. 从Root开始
2. 检查MBR1(0,0,8,10)与查询窗口相交 ✓
3. 递归检查B(4,4,8,8), 相交 ✓ → 返回Park
4. 检查MBR2(8,0,15,10), 不相交 ✗ → 剪枝
5. 避免检查整个C区域, 查询加速!

6.3.3 空间分区策略

KDB-Tree分区示例：

原始数据分布 (100万个点):
    ●●●●●●●●●●●●●●●●●●●●
    ●●●●●●●●●●●●●●●●●●●●
    ●●●●●●●         ●●●●●
    ●●●●●●●         ●●●●●
    ●●●●●●●●●●●●●●●●●●●●

KDB-Tree分区过程:
步骤1: 对X轴排序, 中位数分割
    |          |          |
    ├──────────┼──────────┤
    | 区域1     | 区域2    | 区域3
    | 33%数据   | 34%数据  | 33%数据
    
步骤2: 对每个区域按Y轴再分割
    ┌──────┬──────┬──────┐
    │ 1.1  │ 2.1  │ 3.1  │ 各11%
    ├──────┼──────┼──────┤
    │ 1.2  │ 2.2  │ 3.2  │ 各11%
    ├──────┼──────┼──────┤
    │ 1.3  │ 2.3  │ 3.3  │ 各11%
    └──────┴──────┴──────┘

最终: 9个分区, 每个约11万个点, 负载均衡!

优势:
✓ 自适应数据分布
✓ 负载均衡
✓ 空间连接时减少跨分区通信

6.3.4 Shuffle机制详解

空间连接的Shuffle过程:

假设: 1000万点 Join 1万个区域

[Before Shuffle]
Executor1: 点1-250万, 区域1-2500
Executor2: 点251-500万, 区域2501-5000
Executor3: 点501-750万, 区域5001-7500
Executor4: 点751-1000万, 区域7501-10000

[Shuffle Phase]
根据空间分区键重新分布数据

Executor1 → 发送数据到各Executor
├─ 分区1的点 → Executor1
├─ 分区2的点 → Executor2
├─ 分区3的点 → Executor3
└─ 分区4的点 → Executor4

[After Shuffle]
Executor1: 分区1的所有点和区域 (本地Join)
Executor2: 分区2的所有点和区域 (本地Join)
Executor3: 分区3的所有点和区域 (本地Join)
Executor4: 分区4的所有点和区域 (本地Join)

Shuffle开销:
- 网络传输: ~GB级数据
- 序列化/反序列化: CPU密集
- 磁盘I/O: 如果内存不足会spill到磁盘

优化策略:
✓ 增加分区数减少单个分区大小
✓ 使用Kryo序列化加速
✓ 增加shuffle内存配置
✓ 使用SSD加速spill操作

6.4 实例分析

实例1: 统计每个城市区域的POI数量

业务场景：
有1000万个POI点（商店、餐厅等）和500个城市区域多边形，需要统计每个区域内的POI数量。

完整代码与执行流程：

from pyspark.sql import SparkSession
from sedona.spark import SedonaContext
from sedona.sql import st_functions as STF
from sedona.core.enums import GridType, IndexType

# ============ 步骤1: 初始化Spark和Sedona ============
spark = SparkSession.builder \
    .appName("POI Analysis") \
    .master("spark://master:7077") \
    .config("spark.executor.instances", "10") \
    .config("spark.executor.cores", "4") \
    .config("spark.executor.memory", "8g") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrator", "org.apache.sedona.core.serde.SedonaKryoRegistrator") \
    .getOrCreate()

sedona = SedonaContext.create(spark)

print("✓ Spark集群连接成功")
# → 此时Python通过Py4J与JVM建立连接
# → Spark Driver向YARN申请10个Executor资源
# → 每个Executor分配4核8GB内存

# ============ 步骤2: 加载POI数据 ============
poi_df = spark.read \
    .option("header", "true") \
    .csv("hdfs://namenode:9000/data/poi.csv")

print(f"✓ POI数据加载完成: {poi_df.count()}条记录")
# → Spark Driver连接HDFS NameNode
# → NameNode返回poi.csv的数据块位置
# → Driver创建读取任务分配给10个Executor
# → 每个Executor读取约100万条POI记录

# ============ 步骤3: 创建空间几何对象 ============
poi_df = poi_df.withColumn(
    "geometry",
    STF.ST_Point(poi_df.longitude.cast("double"), poi_df.latitude.cast("double"))
)

print("✓ POI空间对象创建完成")
# → 在每个Executor中执行
# → Sedona将经纬度坐标转换为JTS Point对象
# → 每个Point对象占用约48字节内存

# ============ 步骤4: 加载区域多边形数据 ============
region_df = spark.read \
    .format("json") \
    .load("hdfs://namenode:9000/data/regions.json")

region_df = region_df.withColumn(
    "geometry",
    STF.ST_GeomFromWKT(region_df.wkt_geometry)
)

print(f"✓ 区域数据加载完成: {region_df.count()}个区域")
# → 500个多边形对象
# → 每个多边形平均包含100-1000个顶点
# → 总大小约5MB (可以broadcast)

# ============ 步骤5: 空间连接 ============
# 使用Broadcast Join（因为region_df较小）
result_df = poi_df.join(
    region_df.hint("broadcast"),
    STF.ST_Contains(region_df.geometry, poi_df.geometry),
    "inner"
)

print("✓ 空间连接执行中...")
# → Driver将region_df广播到所有Executor（5MB）
# → 各Executor本地执行空间包含测试
# → 对于每个POI点:
#    1. 遍历500个区域多边形
#    2. 使用JTS算法测试点是否在多边形内
#    3. Ray Casting算法：从点发射射线，计算与多边形边的交点数
#    4. 如果交点数为奇数，点在多边形内
# → 无需Shuffle，性能最优

# ============ 步骤6: 聚合统计 ============
summary_df = result_df.groupBy("region_id", "region_name") \
    .agg(
        count("*").alias("poi_count"),
        collect_list("poi_name").alias("poi_list")
    ) \
    .orderBy("poi_count", ascending=False)

print("✓ 统计计算完成")
# → 各Executor本地预聚合
# → Shuffle到少数Executor进行最终聚合
# → 返回500行结果（每个区域一行）

# ============ 步骤7: 结果展示 ============
summary_df.show(10)
# → Driver通过Py4J将结果返回Python
# → 在控制台显示前10个区域统计

# 保存结果到HDFS
summary_df.write \
    .mode("overwrite") \
    .parquet("hdfs://namenode:9000/output/poi_summary")

print("✅ 分析完成！结果已保存到HDFS")

spark.stop()

执行时间分析：

内存使用分析：

单机模式:
- POI数据: 1000万 × 48字节 ≈ 480MB
- 区域数据: 5MB
- 中间结果: 约2GB
- 总需求: ~3GB (可运行但较慢)

集群模式:
- 每个Executor处理100万POI: 48MB
- 广播区域数据: 5MB
- 每Executor内存使用: ~200MB
- 剩余内存用于缓存和Shuffle

实例2: 查找最近的充电站

业务场景：
为每辆电动汽车（10万辆）查找最近的充电站（1万个）。

完整代码：

from pyspark.sql import SparkSession
from sedona.spark import SedonaContext
from sedona.sql import st_functions as STF
from pyspark.sql.functions import row_number
from pyspark.sql.window import Window

# 初始化
spark = SparkSession.builder \
    .appName("Nearest Charger") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrator", "org.apache.sedona.core.serde.SedonaKryoRegistrator") \
    .getOrCreate()

sedona = SedonaContext.create(spark)

# 加载电动车位置
vehicles_df = spark.read.csv("hdfs://namenode:9000/data/vehicles.csv", header=True)
vehicles_df = vehicles_df.withColumn(
    "geometry",
    STF.ST_Point(vehicles_df.lon.cast("double"), vehicles_df.lat.cast("double"))
)

# 加载充电站位置
chargers_df = spark.read.csv("hdfs://namenode:9000/data/chargers.csv", header=True)
chargers_df = chargers_df.withColumn(
    "geometry",
    STF.ST_Point(chargers_df.lon.cast("double"), chargers_df.lat.cast("double"))
)

# 使用Broadcast Join + 距离计算
result_df = vehicles_df.crossJoin(chargers_df.hint("broadcast")) \
    .withColumn(
        "distance_km",
        STF.ST_Distance(vehicles_df.geometry, chargers_df.geometry) / 1000
    )

# 为每辆车选择最近的充电站
window = Window.partitionBy("vehicle_id").orderBy("distance_km")
nearest_df = result_df.withColumn("rank", row_number().over(window)) \
    .filter("rank = 1") \
    .select(
        "vehicle_id",
        "vehicle_name",
        "charger_id",
        "charger_name",
        "distance_km"
    )

nearest_df.show(20)

# 统计距离分布
nearest_df.groupBy(
    (nearest_df.distance_km / 5).cast("int").alias("distance_range")
).count().orderBy("distance_range").show()

spark.stop()

技术细节：

• Cross Join: 生成10万×1万=10亿行中间结果
• Broadcast: 1万个充电站数据（~1MB）广播到所有Executor
• 距离计算: 使用Haversine公式计算地球表面距离
• Window函数: 为每辆车分区，按距离排序取第一个

7. 应用示例

7.1 基础空间查询

示例1: 范围查询

from sedona.sql import st_functions as STF

# 查询北京市范围内的所有POI
beijing_bbox = "POLYGON((115.7 39.4, 117.4 39.4, 117.4 41.6, 115.7 41.6, 115.7 39.4))"

result = poi_df.filter(
    STF.ST_Within(
        poi_df.geometry,
        STF.ST_GeomFromText(beijing_bbox)
    )
)

result.show()

示例2: 缓冲区分析

# 创建500米缓冲区
buffered_df = poi_df.withColumn(
    "buffer_500m",
    STF.ST_Buffer(poi_df.geometry, 500)  # 单位：米
)

# 查找缓冲区内的其他POI
nearby_pois = buffered_df.alias("a").join(
    poi_df.alias("b"),
    STF.ST_Intersects(
        buffered_df.buffer_500m,
        poi_df.geometry
    ) & (buffered_df.poi_id != poi_df.poi_id)
)

7.2 空间连接操作

示例1: 空间包含查询

# 查找每个行政区的学校数量
schools_per_district = districts_df.join(
    schools_df,
    STF.ST_Contains(districts_df.geometry, schools_df.geometry)
).groupBy("district_id", "district_name") \
 .agg(count("*").alias("school_count"))

示例2: 空间相交分析

# 查找与河流相交的道路
roads_crossing_rivers = roads_df.join(
    rivers_df,
    STF.ST_Intersects(roads_df.geometry, rivers_df.geometry)
).select(
    roads_df.road_name,
    rivers_df.river_name,
    STF.ST_Intersection(roads_df.geometry, rivers_df.geometry).alias("crossing_point")
)

7.3 性能优化技巧

优化1: 合理设置分区数

# 根据集群规模设置分区数
num_executors = 10
cores_per_executor = 4
partition_count = num_executors * cores_per_executor * 3  # 120个分区

df = df.repartition(partition_count)

优化2: 使用持久化缓存

# 对频繁使用的数据进行缓存
poi_df.cache()
poi_df.count()  # 触发缓存

# 使用完毕后释放
poi_df.unpersist()

优化3: 空间分区优化

from sedona.core.enums import GridType

# 对大数据集使用KDB-Tree分区
spatial_rdd.spatialPartitioning(GridType.KDBTREE, 200)
spatial_rdd.spatialPartitionedRDD.cache()

优化4: 谓词下推

# 先过滤再连接，减少数据量
filtered_poi = poi_df.filter(poi_df.category == "restaurant")
result = filtered_poi.join(districts_df, ...)

8. 常见问题与解决方案

问题1: 内存溢出 (OOM)

现象：

java.lang.OutOfMemoryError: Java heap space

解决方案：

# 增加Executor内存
spark.executor.memory=16g
spark.executor.memoryOverhead=4g

# 增加Driver内存
spark.driver.memory=8g

# 增加分区数，减少单个分区大小
spark.sql.shuffle.partitions=400

问题2: Shuffle性能慢

现象：
Shuffle阶段耗时过长，大量磁盘I/O

解决方案：

# 增加Shuffle内存
spark.shuffle.memoryFraction=0.4

# 使用SSD存储Shuffle数据
spark.local.dir=/ssd/spark-tmp

# 启用Shuffle服务
spark.shuffle.service.enabled=true

问题3: 数据倾斜

现象：
某些Task执行时间远超其他Task

解决方案：

# 1. 重新分区
df = df.repartition(200, "partition_key")

# 2. 添加随机前缀
from pyspark.sql.functions import rand, concat, lit
df = df.withColumn("salt", (rand() * 10).cast("int"))
df = df.repartition("salt", "original_key")

# 3. 使用Broadcast Join避免Shuffle
small_df.hint("broadcast")

问题4: Sedona函数不可用

现象：

AnalysisException: Undefined function: ST_Point

解决方案：

# 确保正确初始化SedonaContext
from sedona.spark import SedonaContext
sedona = SedonaContext.create(spark)

# 或手动注册函数
from sedona.register import SedonaRegistrator
SedonaRegistrator.registerAll(spark)

问题5: HDFS连接失败

现象：

java.io.IOException: Failed to connect to HDFS

解决方案：

# 检查HADOOP_CONF_DIR环境变量
echo $HADOOP_CONF_DIR

# 确保core-site.xml和hdfs-site.xml配置正确
vim $HADOOP_CONF_DIR/core-site.xml

# 检查HDFS服务状态
hdfs dfsadmin -report

9. 附录

9.1 常用Sedona函数速查表

9.2 性能调优参数汇总

# ========== 内存配置 ==========
spark.driver.memory=8g
spark.driver.memoryOverhead=2g
spark.executor.memory=16g
spark.executor.memoryOverhead=4g
spark.memory.fraction=0.8
spark.memory.storageFraction=0.3

# ========== 并行度配置 ==========
spark.executor.instances=20
spark.executor.cores=4
spark.sql.shuffle.partitions=200
spark.default.parallelism=200

# ========== 序列化配置 ==========
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrator=org.apache.sedona.core.serde.SedonaKryoRegistrator
spark.kryoserializer.buffer.max=256m

# ========== Shuffle优化 ==========
spark.shuffle.service.enabled=true
spark.shuffle.memoryFraction=0.4
spark.reducer.maxSizeInFlight=96m
spark.shuffle.io.maxRetries=5

# ========== 网络配置 ==========
spark.network.timeout=600s
spark.rpc.askTimeout=300s
spark.locality.wait=3s

# ========== Sedona特定配置 ==========
sedona.join.numpartition=200
sedona.join.gridtype=kdbtree
sedona.join.indexbuildside=left

9.3 参考资源

官方文档：

• Apache Sedona: https://sedona.apache.org/
• Apache Spark: https://spark.apache.org/docs/latest/
• Hadoop HDFS: https://hadoop.apache.org/docs/stable/

社区资源：

• GitHub: https://github.com/apache/sedona
• Stack Overflow: 标签 apache-sedona
• 邮件列表: dev@sedona.apache.org

学习资源：

• Sedona Tutorial: https://sedona.apache.org/tutorial/
• Spark SQL Guide: https://spark.apache.org/docs/latest/sql-programming-guide.html
• GeoSpatial Analysis with Python: 相关书籍和课程

结语

本文档详细介绍了Apache Sedona的部署、集群化配置以及技术原理。通过本文档，你应该能够：

✅ 完成单机部署 - 用于开发和测试
✅ 实施集群部署 - 用于生产环境
✅ 理解技术原理 - 掌握Python-Spark-Hadoop-Sedona的交互机制
✅ 优化性能 - 根据业务场景调优配置
✅ 解决常见问题 - 快速排查和修复错误

核心要点回顾：

1. 单机 vs 集群的核心区别

   • 执行模式：Local vs Distributed
   • 数据存储：本地文件 vs HDFS
   • 性能差异：10-50倍提升
   • 运维复杂度：简单 vs 复杂

2. 技术栈交互流程

   • Python → Py4J → Spark Driver → Executor → HDFS
   • 13个步骤完整生命周期
   • 空间分区、索引、Shuffle机制

3. 性能优化关键

   • 合理配置内存和分区数
   • 使用空间索引和分区
   • 选择合适的Join策略
   • 监控和调优

下一步建议：

• 在测试环境验证配置
• 进行性能基准测试
• 逐步扩展到生产环境
• 持续监控和优化

如有问题，欢迎参考官方文档或联系社区支持。

10. 生产环境部署操作规范

10.1 部署前准备工作

10.1.1 现状分析与评估

1. 业务现状调研

调研清单：
┌──────────────────────────────────────────────────────┐
│ □ 当前空间数据处理方式（数据库、单机脚本等）           │
│ □ 数据量级和增长趋势（TB级？增长率？）                 │
│ □ 查询响应时间要求（秒级？分钟级？）                   │
│ □ 并发用户数和访问模式                                │
│ □ 业务高峰期时间段                                    │
│ □ 数据更新频率（实时？批处理？）                       │
│ □ 可接受的服务中断时间窗口                            │
└──────────────────────────────────────────────────────┘

2. 资源现状盘点

资源盘点模板：

# 服务器资源清单
服务器清单:
  Master节点:
    - 主机名: spark-master-01
      IP: 192.168.1.10
      CPU: 16核
      内存: 64GB
      磁盘: 500GB SSD + 2TB HDD
      用途: Spark Master + HDFS NameNode
      
  Worker节点:
    - 主机名: spark-worker-01
      IP: 192.168.1.11
      CPU: 32核
      内存: 128GB
      磁盘: 4TB HDD
      用途: Spark Worker + HDFS DataNode
      
网络规划:
  内网带宽: 10Gbps
  公网带宽: 100Mbps
  安全组: spark-cluster-sg
  
存储规划:
  HDFS总容量: 40TB (10节点 × 4TB)
  副本数: 3
  可用容量: 13TB
  预留空间: 20%

10.1.2 数据备份与迁移策略

1. 现有数据备份方案

# 备份计划清单
备份对象:
  □ 业务数据库（PostGIS/MySQL等）
  □ 现有空间数据文件
  □ 应用配置文件
  □ 历史日志文件
  
备份方式:
  □ 全量备份 + 增量备份
  □ 异地备份（至少2个副本）
  □ 备份验证（定期恢复测试）

数据备份脚本示例：

#!/bin/bash
# backup_before_migration.sh

BACKUP_DIR="/backup/pre-sedona-migration/$(date +%Y%m%d)"
mkdir -p $BACKUP_DIR

# 1. 备份PostGIS数据库
echo "备份PostGIS数据库..."
pg_dump -h localhost -U postgres spatial_db > $BACKUP_DIR/spatial_db_backup.sql

# 2. 备份空间数据文件
echo "备份空间数据文件..."
tar -czf $BACKUP_DIR/spatial_data.tar.gz /data/spatial/

# 3. 备份应用配置
echo "备份应用配置..."
cp -r /etc/app_config $BACKUP_DIR/

# 4. 生成备份清单
echo "生成备份清单..."
cat > $BACKUP_DIR/backup_manifest.txt <<EOF
备份时间: $(date)
备份内容:
- 数据库备份: spatial_db_backup.sql ($(du -h $BACKUP_DIR/spatial_db_backup.sql | cut -f1))
- 数据文件: spatial_data.tar.gz ($(du -h $BACKUP_DIR/spatial_data.tar.gz | cut -f1))
- 配置文件: app_config/
备份完成: 是
验证状态: 待验证
EOF

# 5. 备份验证
echo "验证备份完整性..."
md5sum $BACKUP_DIR/* > $BACKUP_DIR/checksums.md5

echo "✅ 备份完成！备份位置: $BACKUP_DIR"

2. 数据迁移方案

迁移策略选择:
┌────────────────────────────────────────────────────┐
│ 方案A: 并行运行（推荐）                              │
│ ├─ 新旧系统同时运行2-4周                            │
│ ├─ 逐步迁移数据和业务                              │
│ └─ 风险：成本增加，维护复杂                         │
│                                                    │
│ 方案B: 停机迁移                                    │
│ ├─ 选择业务低峰期一次性迁移                         │
│ ├─ 迁移时间窗口: 4-8小时                           │
│ └─ 风险：服务中断，回滚困难                         │
│                                                    │
│ 方案C: 分阶段迁移（推荐）                           │
│ ├─ 按业务模块逐步迁移                              │
│ ├─ 每个模块独立测试验证                            │
│ └─ 风险：迁移周期长，需要数据同步                   │
└────────────────────────────────────────────────────┘

数据迁移检查清单：

□ 迁移前数据量统计（行数、文件大小）
□ 迁移后数据量验证（确保无数据丢失）
□ 数据完整性检查（MD5、记录数对比）
□ 数据格式转换验证（WKT/GeoJSON等）
□ 索引重建和性能测试
□ 业务功能回归测试

10.1.3 风险评估与应对

风险矩阵：

详细风险应对方案：

风险1：数据丢失或损坏

预防措施:
✓ 3副本HDFS存储
✓ 跨机架部署
✓ 定期备份到异地
✓ 数据校验机制

应急预案:
1. 立即停止写入操作
2. 从最近备份恢复
3. 数据一致性检查
4. 业务功能验证
5. 事故报告和改进

恢复时间目标(RTO): 2小时
恢复点目标(RPO): 1小时

风险2：性能不达标

预防措施:
✓ 生产前性能测试
✓ 逐步放量验证
✓ 性能监控告警
✓ 资源弹性扩展

应急预案:
1. 快速扩容Worker节点
2. 调整资源分配参数
3. 优化查询执行计划
4. 启用缓存机制
5. 必要时启用降级方案

性能指标:
- 查询响应时间 < 5秒 (P95)
- 并发处理能力 > 100 QPS
- CPU使用率 < 80%
- 内存使用率 < 85%

风险3：集群节点故障

预防措施:
✓ Master节点高可用(ZooKeeper)
✓ HDFS NameNode HA
✓ 自动故障转移
✓ 节点健康检查

应急预案:
1. 自动故障检测(30秒)
2. 自动任务重试
3. 故障节点隔离
4. 通知运维人员
5. 更换故障硬件

恢复策略:
- 单节点故障: 自动恢复
- 多节点故障: 人工介入
- 灾难性故障: 切换备用集群

10.1.4 部署决策与审批

部署决策会议议程：

会议议程:
1. 项目背景和目标 (10分钟)
   - 当前痛点
   - 预期收益
   
2. 技术方案说明 (20分钟)
   - 架构设计
   - 技术栈选择
   - 关键技术点
   
3. 资源需求报告 (10分钟)
   - 服务器资源
   - 存储资源
   - 网络资源
   - 人力投入
   
4. 风险评估报告 (15分钟)
   - 主要风险
   - 应对措施
   - 应急预案
   
5. 实施计划 (10分钟)
   - 时间表
   - 里程碑
   - 验收标准
   
6. 预算评审 (10分钟)
   - 硬件成本
   - 软件成本
   - 人力成本
   
7. 讨论与决策 (15分钟)

部署审批表模板：

┌─────────────────────────────────────────────┐
│        Apache Sedona集群部署审批表            │
├─────────────────────────────────────────────┤
│ 项目名称: 空间数据处理平台升级                 │
│ 申请部门: 数据中心                            │
│ 申请时间: 2025-12-09                         │
│                                              │
│ 资源需求:                                    │
│ □ 服务器: 10台 (规格: 32核/128GB)            │
│ □ 存储: 40TB HDFS                            │
│ □ 网络: 10Gbps内网                           │
│                                              │
│ 预算: 150万元                                │
│                                              │
│ 部署周期: 3个月                              │
│                                              │
│ 风险评估: 已完成 ✓                           │
│ 备份方案: 已制定 ✓                           │
│ 应急预案: 已制定 ✓                           │
│                                              │
│ 审批意见:                                    │
│ _______________________________________________│
│                                              │
│ 技术负责人签字: __________ 日期: __________   │
│ 部门经理签字: __________ 日期: __________     │
│ IT总监签字: __________ 日期: __________       │
└─────────────────────────────────────────────┘

10.1.5 沟通协调与培训

1. 相关方沟通清单

2. 培训计划

培训体系:

管理层培训 (2小时)
├─ 项目价值和收益
├─ 关键风险和应对
└─ 决策支持

技术人员培训 (2天)
├─ Day 1: 基础培训
│   ├─ Spark基础概念
│   ├─ Sedona功能介绍
│   ├─ 集群架构说明
│   └─ 基本操作演示
│
└─ Day 2: 高级培训
    ├─ 性能调优技巧
    ├─ 故障排查方法
    ├─ 监控告警配置
    └─ 实战案例分析

业务用户培训 (4小时)
├─ 功能对比说明
├─ 使用方式变化
├─ 常见问题解答
└─ 反馈渠道说明

运维人员培训 (3天)
├─ Day 1: 部署维护
├─ Day 2: 监控告警
└─ Day 3: 应急演练

培训材料清单：

□ PPT演示文稿
□ 操作手册（中文版）
□ 视频教程
□ 快速参考卡片
□ 常见问题FAQ
□ 技术支持联系方式

10.2 关键配置参数详解

10.2.1 必须配置的关键参数

级别：🔴 极其重要（不配置会导致系统无法运行）

# ========== 序列化配置（Sedona必需）==========
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrator=org.apache.sedona.core.serde.SedonaKryoRegistrator
# 说明: Sedona空间对象序列化必需配置
# 影响: 不配置会导致空间函数不可用
# 调优: 无需调优

# ========== 内存配置 ==========
spark.executor.memory=8g
# 说明: 每个Executor的堆内存大小
# 影响: 直接影响数据处理能力
# 调优: 根据节点内存和Executor数量计算
#       公式: (节点内存 × 0.75) / Executor数量

spark.executor.memoryOverhead=2g
# 说明: Executor堆外内存（用于网络传输、本地缓存等）
# 影响: 防止容器被杀死
# 调优: 建议为executor.memory的20-25%
#       最小值: 384MB

spark.driver.memory=4g
# 说明: Driver进程内存大小
# 影响: 影响任务调度和结果收集
# 调优: 建议为集群总内存的5-10%

# ========== 核心数配置 ==========
spark.executor.cores=4
# 说明: 每个Executor使用的CPU核心数
# 影响: 影响并行任务数量
# 调优: 建议2-5核，不宜过大
#       过大会导致任务竞争CPU

级别：🟡 重要（影响性能和稳定性）

# ========== Executor数量配置 ==========
spark.executor.instances=10
# 说明: 集群中Executor的数量
# 影响: 直接影响并行度
# 调优: 根据集群规模动态调整
#       小集群: 3-10个
#       中集群: 10-50个
#       大集群: 50-200个

# ========== 动态资源分配 ==========
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2
spark.dynamicAllocation.maxExecutors=20
spark.dynamicAllocation.initialExecutors=5
# 说明: 根据负载自动调整Executor数量
# 影响: 提高资源利用率
# 调优: 根据业务负载模式设置
#       稳定负载: 关闭动态分配
#       波动负载: 开启动态分配

# ========== Shuffle分区数 ==========
spark.sql.shuffle.partitions=200
# 说明: Shuffle操作的默认分区数
# 影响: 影响Shuffle性能和并行度
# 调优: 建议为Executor总核心数的2-3倍
#       公式: (Executor数 × cores) × 2.5
#       数据量大: 增加分区数
#       数据量小: 减少分区数

级别：🟢 建议配置（优化性能）

# ========== 网络配置 ==========
spark.network.timeout=600s
# 说明: 网络操作超时时间
# 影响: 防止大数据传输超时
# 调优: 默认120s，大数据量建议300-600s

spark.rpc.askTimeout=300s
# 说明: RPC通信超时时间
# 影响: Driver与Executor通信
# 调优: 默认120s，复杂任务建议增加

# ========== Shuffle优化 ==========
spark.shuffle.service.enabled=true
# 说明: 启用外部Shuffle服务
# 影响: Executor下线后数据仍可访问
# 调优: 生产环境强烈建议开启

spark.reducer.maxSizeInFlight=96m
# 说明: Reducer拉取数据的缓冲区大小
# 影响: Shuffle读取性能
# 调优: 默认48m，网络好时可增加

# ========== 序列化缓冲区 ==========
spark.kryoserializer.buffer.max=256m
# 说明: Kryo序列化缓冲区最大值
# 影响: 大对象序列化
# 调优: 默认64m，大对象场景建议增加

10.2.2 Sedona特定配置参数

# ========== 空间分区配置 ==========
sedona.join.numpartition=200
# 说明: 空间连接的分区数量
# 影响: 空间连接性能
# 调优: 与spark.sql.shuffle.partitions保持一致

sedona.join.gridtype=kdbtree
# 说明: 空间分区策略
# 可选值: kdbtree, quadtree, rtree, voronoi
# 影响: 数据分布均衡性
# 调优: 
#   - kdbtree: 数据分布不均匀（推荐）
#   - quadtree: 数据分布均匀
#   - rtree: 小数据集
#   - voronoi: 特殊场景

sedona.join.indexbuildside=left
# 说明: 为哪一侧数据构建索引
# 可选值: left, right
# 影响: Join性能
# 调优: 为较小的数据集构建索引

# ========== 空间索引配置 ==========
sedona.global.index=true
# 说明: 是否启用全局空间索引
# 影响: 查询性能
# 调优: 大数据集建议启用

sedona.global.indextype=rtree
# 说明: 全局索引类型
# 可选值: rtree, quadtree
# 影响: 索引效率
# 调优: rtree适合大多数场景

10.2.3 根据场景的参数优化

场景1：大数据量批处理（TB级数据）

# 内存配置 - 增大
spark.executor.memory=32g
spark.executor.memoryOverhead=8g
spark.driver.memory=16g

# 并行度配置 - 增大
spark.executor.instances=50
spark.sql.shuffle.partitions=1000
spark.default.parallelism=1000

# Shuffle优化 - 增大缓冲区
spark.reducer.maxSizeInFlight=256m
spark.shuffle.io.maxRetries=10

# 内存管理 - 增加存储比例
spark.memory.fraction=0.8
spark.memory.storageFraction=0.5

# Sedona配置
sedona.join.numpartition=1000
sedona.join.gridtype=kdbtree

场景2：实时/交互式查询（秒级响应）

# 内存配置 - 适中
spark.executor.memory=16g
spark.executor.memoryOverhead=4g
spark.driver.memory=8g

# 并行度配置 - 适中
spark.executor.instances=20
spark.sql.shuffle.partitions=400
spark.default.parallelism=400

# 缓存优化 - 增加缓存比例
spark.memory.fraction=0.8
spark.memory.storageFraction=0.4

# 动态资源分配 - 快速响应
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=5
spark.dynamicAllocation.maxExecutors=30
spark.dynamicAllocation.schedulerBacklogTimeout=1s

# 推测执行 - 减少长尾任务
spark.speculation=true
spark.speculation.multiplier=2

场景3：多用户共享集群

# 资源隔离
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.shuffleTracking.enabled=true

# 公平调度
spark.scheduler.mode=FAIR
spark.scheduler.allocation.file=/path/to/fairscheduler.xml

# 队列配置示例（fairscheduler.xml）
<!-- 
<allocations>
  <pool name="production">
    <schedulingMode>FAIR</schedulingMode>
    <weight>3</weight>
    <minShare>5</minShare>
  </pool>
  <pool name="development">
    <schedulingMode>FAIR</schedulingMode>
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
</allocations>
-->

# 资源限制
spark.executor.memory=8g
spark.executor.cores=4
spark.executor.instances=10

10.2.4 通用配置参数模板

标准生产环境配置（推荐）：

# ========================================
# Spark Sedona生产环境标准配置
# 集群规模: 10节点 × 32核/128GB
# 适用场景: 通用空间数据处理
# ========================================

# ========== 应用基础配置 ==========
spark.app.name=sedona-production
spark.master=spark://master:7077

# ========== 资源配置 ==========
spark.executor.instances=20
spark.executor.cores=4
spark.executor.memory=16g
spark.executor.memoryOverhead=4g
spark.driver.memory=8g
spark.driver.memoryOverhead=2g

# ========== 动态资源分配 ==========
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=5
spark.dynamicAllocation.maxExecutors=30
spark.dynamicAllocation.initialExecutors=10

# ========== 序列化配置 ==========
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrator=org.apache.sedona.core.serde.SedonaKryoRegistrator
spark.kryoserializer.buffer.max=256m

# ========== 并行度配置 ==========
spark.sql.shuffle.partitions=200
spark.default.parallelism=200

# ========== 内存管理 ==========
spark.memory.fraction=0.8
spark.memory.storageFraction=0.3

# ========== Shuffle配置 ==========
spark.shuffle.service.enabled=true
spark.shuffle.memoryFraction=0.4
spark.reducer.maxSizeInFlight=96m
spark.shuffle.io.maxRetries=5

# ========== 网络配置 ==========
spark.network.timeout=600s
spark.rpc.askTimeout=300s
spark.locality.wait=3s

# ========== 日志和监控 ==========
spark.eventLog.enabled=true
spark.eventLog.dir=hdfs://namenode:9000/spark-logs
spark.history.fs.logDirectory=hdfs://namenode:9000/spark-logs

# ========== Sedona配置 ==========
sedona.join.numpartition=200
sedona.join.gridtype=kdbtree
sedona.join.indexbuildside=left
sedona.global.index=true
sedona.global.indextype=rtree

# ========== 安全配置 ==========
spark.authenticate=true
spark.network.crypto.enabled=true

10.2.5 参数调优监控指标

需要持续监控的关键指标：

CPU相关:
├─ CPU使用率 < 80% (平均值)
├─ CPU等待率 < 10%
└─ 负载均衡度 (各节点CPU使用率方差 < 15%)

内存相关:
├─ 堆内存使用率 < 85%
├─ 堆外内存使用率 < 80%
├─ GC时间占比 < 10%
└─ Full GC频率 < 1次/小时

磁盘相关:
├─ 磁盘使用率 < 80%
├─ 磁盘I/O等待 < 20%
└─ HDFS副本完整性 = 100%

网络相关:
├─ 网络带宽使用率 < 70%
├─ Shuffle读写速度 > 100MB/s
└─ 网络延迟 < 5ms (内网)

任务相关:
├─ 任务失败率 < 1%
├─ 数据倾斜度 < 3 (最大任务时间/平均任务时间)
└─ Shuffle倾斜度 < 3

参数调优决策流程：

问题: 任务执行慢

Step 1: 查看Spark UI
├─ Stage耗时分布
├─ Task执行时间
└─ Shuffle数据量

Step 2: 定位瓶颈
├─ CPU瓶颈?
│   └─ 增加executor.cores或instances
├─ 内存瓶颈?
│   └─ 增加executor.memory
├─ Shuffle瓶颈?
│   └─ 增加shuffle.partitions
└─ 数据倾斜?
    └─ 调整分区策略或添加salt

Step 3: 调整参数
├─ 单个参数每次调整20-30%
├─ 观察效果1-2天
└─ 记录调优日志

Step 4: 验证效果
├─ 对比调优前后性能
├─ 检查副作用
└─ 文档记录调优过程

调优日志模板：

## 调优记录

**日期：** 2025-12-09  
**调优人员：** 张三  
**问题描述：** 空间连接任务执行缓慢，平均耗时45分钟

### 性能分析
- Spark UI观察：Shuffle阶段耗时占比70%
- Shuffle数据量：500GB
- 分区数：200
- 数据倾斜：最大任务时间是平均时间的5倍

### 调优措施
1. 增加Shuffle分区数：200 → 400
2. 启用推测执行
3. 调整空间分区策略：quadtree → kdbtree

### 调优结果
| 指标 | 调优前 | 调优后 | 改善 |
|------|--------|--------|------|
| 总执行时间 | 45分钟 | 12分钟 | 73% |
| Shuffle时间 | 32分钟 | 6分钟 | 81% |
| 数据倾斜度 | 5.2 | 1.8 | 65% |

### 经验总结
- KDB-Tree分区策略在数据分布不均匀时效果更好
- Shuffle分区数建议为executor总核心数的3-5倍
- 持续监控数据倾斜指标

文档编写： 郭宁泰
最后更新： 2025年12月9日
版本： V2.0（优化版 - 生产环境增强版）