spark RDD常用算子

### Spark RDD 转换算子的使用方法及示例 #### 什么是转换算子？ 在Spark中，RDD（弹性分布式数据集）支持两种主要的操作：**转换操作**和**动作操作**。转换操作是对现有RDD执行的一种操作，返回一个新的RDD。这些操作不会立即触发计算，而是构建一个逻辑计划来描述如何从其他已有的RDD派生新的RDD[^1]。 #### 常见的转换算子及其功能 以下是几个常见的转换算子以及它们的功能： 1. **map(func)** 对RDD中的每一个元素应用函数`func`并返回一个新的RDD。 2. **filter(func)** 返回一个新的RDD，其中只包含满足条件`func(x)`为真的元素。 3. **flatMap(func)** 类似于`map`，但是可以将每个输入项映射到多个输出项。 4. **union(otherDataset)** 返回两个RDD的并集，即包含两个RDD中所有元素的新RDD[^2]。 5. **join(otherDataset, numPartitions)** 执行两组键值对RDD之间的内连接操作。 6. **groupByKey([numTasks])** 当处理由(key, value)组成的RDD时，此操作会针对每个key聚合所有的value。 7. **reduceByKey(func, [numTasks])** 针对具有相同key的所有values调用指定的二元运算符`func`，从而减少该key对应的values数量。 8. **sortByKey(ascending=True, numPartitions=None)** 按照key排序，可以选择升序或者降序排列。 9. **distinct([numTasks])** 删除重复的数据条目，保留唯一的结果集合。 --- #### 示例代码展示 下面是一些具体的例子，展示了上述提到的一些常用转换算子的实际运用方式。 ##### map() 示例 ```python rdd = sc.parallelize([1, 2, 3, 4]) mapped_rdd = rdd.map(lambda x: x * 2) print(mapped_rdd.collect()) # 输出: [2, 4, 6, 8] ``` ##### filter() 示例 ```python filtered_rdd = rdd.filter(lambda x: x % 2 == 0) print(filtered_rdd.collect()) # 输出: [2, 4] ``` ##### flatMap() 示例 ```python words_list = ["hello world", "spark is great"] rdd_words = sc.parallelize(words_list) flattened_rdd = rdd_words.flatMap(lambda line: line.split(" ")) print(flattened_rdd.collect()) # 输出: ['hello', 'world', 'spark', 'is', 'great'] ``` ##### union() 示例 ```python rdd1 = sc.parallelize([1, 2, 3]) rdd2 = sc.parallelize([4, 5, 6]) united_rdd = rdd1.union(rdd2) print(united_rdd.collect()) # 输出: [1, 2, 3, 4, 5, 6] ``` ##### reduceByKey() 示例 ```python pairs = sc.parallelize([('a', 1), ('b', 1), ('a', 1)]) reduced_pairs = pairs.reduceByKey(lambda a, b: a + b) print(reduced_pairs.collect()) # 输出: [('a', 2), ('b', 1)] ``` --- ### 总结 通过以上介绍可以看出，Spark RDD 的转换算子提供了强大的工具用于数据变换和准备阶段的工作。每种转换都有其特定用途，在实际开发过程中可以根据需求灵活选用不同的转换算子组合实现复杂业务逻辑。

### 创建RDD 在Apache Spark中，可以通过并行化集合或读取外部数据源来创建弹性分布式数据集（Resilient Distributed Dataset, RDD）。对于Scala和Python这两种编程语言而言，操作方式有所不同。 #### 使用Scala创建RDD 通过`SparkContext.parallelize()`方法可以将已有的集合转换成RDD。下面是一个简单的例子： ```scala val sc = new org.apache.spark.SparkContext("local", "First App") // 并行化现有数组到集群上形成RDD val data = Array(1, 2, 3, 4, 5) val distData = sc.parallelize(data) ``` 为了从本地文件系统中的文件创建RDD，同样利用`textFile`函数指定路径即可[^1]。 ```scala // 从本地文件创建RDD val logData = sc.textFile("file:///path/to/local/file.txt").cache() ``` #### 使用Python创建RDD PySpark提供了几乎相同的API用于处理相同的操作，在这里展示如何用Python实现上述功能： ```python from pyspark import SparkConf, SparkContext conf = SparkConf().setAppName('FirstApp') sc = SparkContext(conf=conf) # 将列表转化为RDD data = [1, 2, 3, 4, 5] dist_data = sc.parallelize(data) # 从本地文件加载数据作为RDD log_data = sc.textFile("/path/to/local/file.txt").cache() ``` ### 转换与动作运算符的应用 一旦有了RDD之后就可以应用各种各样的转换(`Transformations`)和行动(`Actions`)操作来进行数据分析工作了。 #### Transformations变换 这些是懒惰求值的，意味着它们不会立即执行计算；相反，它们只是记录下要应用于基础数据上的操作序列。常见的有map()、filter()等。 ##### Scala版本: ```scala // 对每个元素乘以2再过滤掉小于等于8的结果 val multipliedFiltered = distData.map(_ * 2).filter(_ > 8) ``` ##### Python版本: ```python # 同样地对每个元素加倍后再筛选大于8的数据项 multiplied_filtered = (dist_data .map(lambda x: x * 2) .filter(lambda y: y > 8)) ``` #### Actions行为 当调用了action类型的命令时才会触发实际的任务提交给集群去运行之前定义好的一系列transformation指令链路。collect(), count()都是常用的actions之一。 ##### Scala实例: ```scala println(multipliedFiltered.collect().mkString(", ")) println(s"Total elements after transformation: ${multipliedFiltered.count()}") ``` ##### Python实例: ```python print(",".join(map(str, multiplied_filtered.collect()))) print(f"Total elements after transformation: {multiplied_filtered.count()}") ``` 以上就是关于怎样使用Scala或者Python编写基本的Spark应用程序来创建RDD、读取本地文件生成RDD以及运用Transformation和Action算子的方法介绍[^4]。