spark行动算子

在 Apache Spark 中，行动算子（Action）用于触发对 RDD 的实际计算，并将结果返回给驱动程序（Driver）或保存到外部存储系统中。与转换算子（Transformation）不同，行动算子会立即触发作业的执行，因为它们需要将计算结果输出到外部。

以下是 Spark 中常见的行动算子及其功能：

返回值到驱动程序的行动算子

1. collect()

   • 功能：将 RDD 中的所有元素返回到驱动程序中，以数组的形式返回。
   • 用途：用于查看 RDD 的内容，但需注意，如果 RDD 很大，可能会导致驱动程序内存溢出。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     result = rdd.collect()
     print(result)  # 输出：[1, 2, 3, 4, 5]
     

2. take(n)

   • 功能：返回 RDD 中的前 n 个元素。
   • 用途：用于快速查看 RDD 的部分数据。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     result = rdd.take(3)
     print(result)  # 输出：[1, 2, 3]
     

3. first()

   • 功能：返回 RDD 中的第一个元素。
   • 用途：用于获取单个元素。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     result = rdd.first()
     print(result)  # 输出：1
     

4. count()

   • 功能：返回 RDD 中的元素个数。
   • 用途：用于统计 RDD 的大小。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     result = rdd.count()
     print(result)  # 输出：5
     

5. countByKey()

   • 功能：对键值对 RDD，返回每个键对应的元素个数。
   • 用途：用于统计每个键的出现次数。
   • 示例：

     rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3)])
     result = rdd.countByKey()
     print(result)  # 输出：{'a': 2, 'b': 1}
     

6. reduce(func)

   • 功能：对 RDD 中的所有元素应用函数 func，并将结果返回到驱动程序。
   • 用途：用于对 RDD 进行聚合操作。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     result = rdd.reduce(lambda a, b: a + b)
     print(result)  # 输出：15
     

将结果保存到外部存储的行动算子

1. saveAsTextFile(path)

   • 功能：将 RDD 保存为文本文件。
   • 用途：用于将 RDD 的内容保存到文件系统（如 HDFS、本地文件系统等）。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     rdd.saveAsTextFile("output/path")
     

2. saveAsSequenceFile(path)

   • 功能：将 RDD 保存为 Hadoop 序列文件。
   • 用途：用于将 RDD 保存为二进制格式的文件。
   • 示例：

     rdd = sc.parallelize([("a", 1), ("b", 2)])
     rdd.saveAsSequenceFile("output/path")
     

3. saveAsObjectFile(path)

   • 功能：将 RDD 保存为序列化对象文件。
   • 用途：用于将 RDD 以 Python 对象的形式保存到文件中。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     rdd.saveAsObjectFile("output/path")
     

4. saveAsParquetFile(path)

   • 功能：将 RDD 保存为 Parquet 文件（仅适用于 DataFrame 或 Dataset）。
   • 用途：用于将数据保存为高效的列存储格式。
   • 示例：

     df = spark.createDataFrame([(1, "a"), (2, "b")], ["id", "value"])
     df.write.parquet("output/path")
     

其他行动算子

1. foreach(func)

   • 功能：对 RDD 中的每个元素应用函数 func，但不会返回结果。
   • 用途：用于对 RDD 的每个元素执行操作，例如写入数据库。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     rdd.foreach(lambda x: print(x))
     

2. takeSample(withReplacement, num, seed)

   • 功能：从 RDD 中随机抽取 num 个样本。
   • 用途：用于获取 RDD 的随机样本。
   • 示例：

     rdd = sc.parallelize([1, 2, 3, 4, 5])
     result = rdd.takeSample(False, 3, seed=42)
     print(result)  # 输出：[2, 4, 5]
     

3. takeOrdered(n, key=None)

   • 功能：返回 RDD 中的前 n 个元素，根据指定的排序函数排序。
   • 用途：用于获取排序后的前 n 个元素。
   • 示例：

     rdd = sc.parallelize([("a", 1), ("b", 2), ("c", 3)])
     result = rdd.takeOrdered(2, key=lambda x: x[1])
     print(result)  # 输出：[('a', 1), ('b', 2)]
     

注意事项

• 性能优化：行动算子会触发作业的执行，因此在使用时需要注意性能优化。例如，避免频繁调用 collect() 或 take()，因为它们会将大量数据返回到驱动程序。
• 资源管理：某些行动算子（如 foreach）可能会对资源使用产生较大影响，尤其是在处理大规模数据时。

行动算子是 Spark 中用于触发实际计算的关键操作，合理使用它们可以高效地完成数据处理任务。