spark RDD transformation操作

2024-05-09 06:32
文章标签 操作 transformation spark rdd

本文主要是介绍spark RDD transformation操作，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

spark RDD transformation操作 
import org.apache.spark.SparkConf
import org.apache.spark.SparkContextobject TransformationOperation {def main(args: Array[String]) {// map()  // filter()  // flatMap()  // groupByKey() // reduceByKey()  // sortByKey() // join()  }def map() {val conf = new SparkConf().setAppName("map").setMaster("local")  val sc = new SparkContext(conf)val numbers = Array(1, 2, 3, 4, 5)val numberRDD = sc.parallelize(numbers, 1)  val multipleNumberRDD = numberRDD.map { num => num * 2 }  multipleNumberRDD.foreach { num => println(num) }   }def filter() {val conf = new SparkConf().setAppName("filter").setMaster("local")val sc = new SparkContext(conf)val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)val numberRDD = sc.parallelize(numbers, 1)val evenNumberRDD = numberRDD.filter { num => num % 2 == 0 }evenNumberRDD.foreach { num => println(num) }   }def flatMap() {val conf = new SparkConf().setAppName("flatMap")  .setMaster("local")  val sc = new SparkContext(conf) val lineArray = Array("hello you", "hello me", "hello world")  val lines = sc.parallelize(lineArray, 1)val words = lines.flatMap { line => line.split(" ") }   words.foreach { word => println(word) }}def groupByKey() {val conf = new SparkConf().setAppName("groupByKey")  .setMaster("local")  val sc = new SparkContext(conf)val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 75),Tuple2("class1", 90), Tuple2("class2", 60))val scores = sc.parallelize(scoreList, 1)  val groupedScores = scores.groupByKey() groupedScores.foreach(score => { println(score._1); score._2.foreach { singleScore => println(singleScore) };println("=============================")  })}def reduceByKey() {val conf = new SparkConf().setAppName("groupByKey")  .setMaster("local")  val sc = new SparkContext(conf)val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 75),Tuple2("class1", 90), Tuple2("class2", 60))val scores = sc.parallelize(scoreList, 1)  val totalScores = scores.reduceByKey(_ + _)  totalScores.foreach(classScore => println(classScore._1 + ": " + classScore._2))  }def sortByKey() {val conf = new SparkConf().setAppName("sortByKey")  .setMaster("local")  val sc = new SparkContext(conf)val scoreList = Array(Tuple2(65, "leo"), Tuple2(50, "tom"), Tuple2(100, "marry"), Tuple2(85, "jack"))  val scores = sc.parallelize(scoreList, 1)  val sortedScores = scores.sortByKey(false)sortedScores.foreach(studentScore => println(studentScore._1 + ": " + studentScore._2))  }def join() {val conf = new SparkConf().setAppName("join")  .setMaster("local")  val sc = new SparkContext(conf)val studentList = Array(Tuple2(1, "leo"),Tuple2(2, "jack"),Tuple2(3, "tom"));val scoreList = Array(Tuple2(1, 100),Tuple2(2, 90),Tuple2(3, 60));val students = sc.parallelize(studentList);val scores = sc.parallelize(scoreList);val studentScores = students.join(scores)  studentScores.foreach(studentScore => { println("student id: " + studentScore._1);println("student name: " + studentScore._2._1)println("student socre: " + studentScore._2._2)  println("=======================================")  })  }def cogroup() {}}


这篇关于spark RDD transformation操作的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/972638。

相关文章

动手学深度学习【数据操作+数据预处理】

动手学深度学习【数据操作+数据预处理】

import osos.makedirs(os.path.join('.', 'data'), exist_ok=True)data_file = os.path.join('.', 'data', 'house_tiny.csv')with open(data_file, 'w') as f:f.write('NumRooms,Alley,Price\n') # 列名f.write('NA
阅读更多...
线程的四种操作

线程的四种操作

所属专栏：Java学习        1. 线程的开启 start和run的区别： run：描述了线程要执行的任务，也可以称为线程的入口 start：调用系统函数，真正的在系统内核中创建线程（创建PCB，加入到链表中），此处的start会根据不同的系统，分别调用不同的api，创建好之后的线程，再单独去执行run（所以说，start的本质是调用系统api，系统的api
阅读更多...
Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering)

Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering)

Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering) Power Iteration Clustering (PIC) 是一种基于图的聚类算法，用于在大规模数据集上进行高效的社区检测。PIC 算法的核心思想是通过迭代图的幂运算来发现数据中的潜在簇。该算法适用于处理大规模图数据，特别是在社交网络分析、推荐系统和生物信息学等领域具有广泛应用。Spa
阅读更多...
Java IO 操作——个人理解

Java IO 操作——个人理解

之前一直Java的IO操作一知半解。今天看到一个便文章觉得很有道理（ 原文章），记录一下。 首先，理解Java的IO操作到底操作的什么内容，过程又是怎么样子。          数据来源的操作： 来源有文件，网络数据。使用File类和Sockets等。这里操作的是数据本身，1,0结构。    File file = new File("path");   字
阅读更多...
MySQL——表操作

MySQL——表操作

目录 一、创建表 二、查看表 2.1 查看表中某成员的数据 2.2 查看整个表中的表成员 2.3 查看创建表时的句柄 三、修改表 alter 3.1 重命名 rename 3.2 新增一列 add 3.3 更改列属性 modify 3.4 更改列名称 change 3.5 删除某列 上一篇博客介绍了库的操作，接下来来看一下表的相关操作。 一、创建表 create
阅读更多...
封装MySQL操作时Where条件语句的组织

封装MySQL操作时Where条件语句的组织

在对数据库进行封装的过程中，条件语句应该是相对难以处理的，毕竟条件语句太过于多样性。 条件语句大致分为以下几种： 1、单一条件，比如：where id = 1; 2、多个条件，相互间关系统一。比如：where id > 10 and age > 20 and score < 60; 3、多个条件，相互间关系不统一。比如：where (id > 10 OR age > 20) AND sco
阅读更多...
PHP7扩展开发之流操作

PHP7扩展开发之流操作

前言 啥是流操作？简单来讲就是对一些文件，网络的IO操作。PHP已经把这些IO操作，封装成流操作。这节，我们将使用PHP扩展实现一个目录遍历的功能。PHP示例代码如下： <?phpfunction list_dir($dir) {if (is_dir($dir) === false) {return;} $dh = opendir($dir);if ($dh == false) {ret
阅读更多...
浙大数据结构：树的定义与操作

浙大数据结构：树的定义与操作

四种遍历 #include<iostream>#include<queue>using namespace std;typedef struct treenode *BinTree;typedef BinTree position;typedef int ElementType;struct treenode{ElementType data;BinTree left;BinTre
阅读更多...
浙大数据结构：04-树7 二叉搜索树的操作集

浙大数据结构：04-树7 二叉搜索树的操作集

这道题答案都在PPT上，所以先学会再写的话并不难。 1、BinTree Insert( BinTree BST, ElementType X ) 递归实现，小就进左子树，大就进右子树。 为空就新建结点插入。 BinTree Insert( BinTree BST, ElementType X ){if(!BST){BST=(BinTree)malloc(sizeof(struct TNo
阅读更多...
hibernate修改数据库已有的对象【简化操作】

hibernate修改数据库已有的对象【简化操作】

陈科肇 直接上代码： /*** 更新新的数据并并未修改旧的数据* @param oldEntity 数据库存在的实体* @param newEntity 更改后的实体* @throws IllegalAccessException * @throws IllegalArgumentException */public void updateNew(T oldEntity,T newEntity
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2