Spark 操作指南

Spark 是一个强大的分布式计算框架，支持多种操作，包括数据处理、转换和分析。以下是常见的 Spark 操作分类及示例。

---

RDD 基本操作

转换操作（Transformations）
RDD 转换操作是惰性执行的，只有在触发行动操作时才会真正执行。

• map：对每个元素应用函数

  rdd = sc.parallelize([1, 2, 3])
  mapped_rdd = rdd.map(lambda x: x * 2)
  

• filter：过滤满足条件的元素

  filtered_rdd = rdd.filter(lambda x: x > 1)
  

• flatMap：先映射后扁平化

  rdd = sc.parallelize(["hello world", "hi"])
  flat_rdd = rdd.flatMap(lambda x: x.split(" "))
  

• reduceByKey：按键聚合

  pair_rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3)])
  reduced_rdd = pair_rdd.reduceByKey(lambda x, y: x + y)
  

行动操作（Actions）
行动操作会触发实际计算并返回结果。

• collect：返回所有元素到驱动节点

  collected_data = rdd.collect()
  

• count：统计元素数量

  count = rdd.count()
  

• take：获取前 N 个元素

  first_two = rdd.take(2)
  

• saveAsTextFile：保存到文本文件

  rdd.saveAsTextFile("output_path")
  

---

DataFrame 操作

DataFrame 是结构化数据的高层抽象，支持 SQL 查询和优化。

• 创建 DataFrame

  from pyspark.sql import SparkSession
  spark = SparkSession.builder.appName("example").getOrCreate()
  df = spark.createDataFrame([(1, "a"), (2, "b")], ["id", "value"])
  

• 选择列

  df.select("id").show()
  

• 过滤数据

  df.filter(df["id"] > 1).show()
  

• 分组聚合

  df.groupBy("value").count().show()
  

• SQL 查询

  df.createOrReplaceTempView("table")
  spark.sql("SELECT * FROM table WHERE id > 1").show()
  

---

数据读写

• 读取 CSV 文件

  df = spark.read.csv("file.csv", header=True, inferSchema=True)
  

• 写入 Parquet 文件

  df.write.parquet("output.parquet")
  

• 读取 JSON 文件

  df_json = spark.read.json("data.json")
  

---

性能优化技巧

• 缓存数据：对重复使用的 RDD 或 DataFrame 调用 cache() 或 persist()。

  df.cache()
  

• 广播变量：减少小数据集在节点间的传输开销。

  broadcast_var = sc.broadcast([1, 2, 3])
  

• 分区调整：根据数据量调整分区数。

  rdd = rdd.repartition(10)
  

---

Spark SQL 示例

# 注册临时表并执行 SQL
df.createOrReplaceTempView("people")
result = spark.sql("SELECT name FROM people WHERE age > 20")
result.show()

通过以上操作，可以高效处理大规模数据并利用 Spark 的分布式能力。