# Spark ETL 概述

Spark ETL (Extract, Transform, Load) 是使用 Apache Spark 进行数据抽取、转换和加载的过程，是现代大数据处理的核心技术之一。

## Spark ETL 的主要优势

1. **分布式处理能力**：Spark可以处理PB级数据，利用集群资源并行处理
2. **内存计算**：通过RDD/DataFrame的内存缓存提高处理速度
3. **统一引擎**：支持批处理和流处理，SQL、机器学习、图计算等多种工作负载
4. **丰富的API**：支持Scala、Java、Python、R等多种语言

## 典型Spark ETL流程

### 1. 数据抽取 (Extract)
```python
# 从不同数据源读取数据
df = spark.read \
    .format("csv") \  # 或jdbc, json, parquet等
    .option("header", "true") \
    .load("hdfs://path/to/file.csv")

# 或从数据库读取
jdbc_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("dbtable", "schema.tablename") \
    .load()
```

### 2. 数据转换 (Transform)
```python
from pyspark.sql import functions as F

# 基本转换
cleaned_df = df.filter(df["age"] > 18) \
    .withColumn("new_col", F.expr("CASE WHEN salary > 5000 THEN 'high' ELSE 'low' END")) \
    .dropDuplicates()

# 聚合操作
aggregated_df = cleaned_df.groupBy("department") \
    .agg(
        F.avg("salary").alias("avg_salary"),
        F.count("*").alias("employee_count")
    )

# 复杂转换 - 使用UDF
def complex_transform(x):
    # 自定义逻辑
    return x * 2

complex_udf = F.udf(complex_transform)
transformed_df = cleaned_df.withColumn("transformed", complex_udf("salary"))
```

### 3. 数据加载 (Load)
```python
# 写入到不同目标
transformed_df.write \
    .mode("overwrite") \  # 或append, ignore, error
    .format("parquet") \
    .save("hdfs://path/to/output")

# 写入到数据库
transformed_df.write \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("dbtable", "schema.output_table") \
    .mode("overwrite") \
    .save()
```

## 性能优化技巧

1. **分区处理**：合理设置分区数，避免数据倾斜
2. **缓存中间结果**：对频繁使用的DataFrame使用`.cache()`
3. **广播变量**：小表join大表时使用广播
4. **并行度调整**：根据集群资源调整`spark.default.parallelism`
5. **文件格式选择**：Parquet/ORC等列式存储格式比CSV更高效
6. **内存管理**：合理配置`spark.executor.memory`和`spark.driver.memory`

## 常见问题解决方案

1. **数据倾斜**：
   - 对倾斜键添加随机前缀
   - 使用`salting`技术
   - 调整reduce分区数

2. **内存不足**：
   - 增加executor内存
   - 减少并行度
   - 使用`persist(StorageLevel.DISK_ONLY)`

3. **小文件问题**：
   - 使用`coalesce()`减少输出文件数
   - 调整`spark.sql.shuffle.partitions`

需要更具体的Spark ETL实现细节或某个特定场景的解决方案吗？