RDD的三种创建方式 快来打我* 2022-09-06 00:08 111阅读 0赞 Spark提供三种创建RDD方式: `集合`、`本地文件`、`HDFS文件` * 使用程序中的集合创建RDD,主要用于进行测试,可以在实际部署到集群运行之前,自己使用集合构造一些测试数据,来测试后面的spark应用程序的流程。 * 使用本地文件创建RDD,主要用于临时性地处理一些存储了大量数据的文件 * 使用HDFS文件创建RDD,是最常用的生产环境的处理方式,主要可以针对HDFS上存储的数据,进 行离线批处理操作 ### 集合 ### 如果要通过集合来创建RDD,需要针对程序中的集合,调用`SparkContext的parallelize()`方法。Spark会将集合中的数据拷贝到集群上,形成一个分布式的数据集合,也就是一个RDD。 相当于,集合中的部分数据会到一个节点上,而另一部分数据会到其它节点上。然后就可以用并行的方式来操作这个分布式数据集合了。 调用parallelize()时,有一个重要的参数可以指定,就是将集合切分成多少个partition。 Spark会为每一个partition运行一个task来进行处理。 Spark默认会根据集群的配置来设置partition的数量。我们也可以在调用parallelize()方法时,传入第二个参数,来设置RDD的partition数量,例如:parallelize(arr, 5) **scala版本** import org.apache.spark.{ SparkConf, SparkContext} /** * 需求:使用集合创建RDD */ object CreateRddByArrayScala { def main(args: Array[String]): Unit = { // 创建SparkContext val conf = new SparkConf() conf.setAppName("CreateRddByArrayScala ") // 设置任务名称 .setMaster("local") // local表示在本地执行 val sc = new SparkContext(conf) // 创建集合 var arr = Array(1, 2, 3, 4, 5) // 基于集合创建RDD val rdd = sc.parallelize(arr) val sum = rdd.reduce(_ + _) println(sum) // 停止SparkContext sc.stop() } } **Java版本** import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function2; import java.util.ArrayList; import java.util.Arrays; import java.util.List; /** * 需求:使用集合创建RDD */ public class CreateRddByArrayJava { public static void main(String[] args) { // 创建SparkContext: SparkConf conf = new SparkConf(); conf.setAppName("CreateRddByArrayJava") .setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); // 创建集合 List<Integer> arr = Arrays.asList(1, 2, 3, 4, 5); JavaRDD<Integer> rdd = sc.parallelize(arr); Integer sum = rdd.reduce(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer i1, Integer i2) throws Exception { return i1 + i2; } }); System.out.println(sum); // 停止sparkContext sc.stop(); } } ### 使用本地文件和HDFS文件 ### 通过`SparkContext的textFile()`方法,可以针对本地文件或HDFS文件创建RDD,RDD中的每个元素就是文件中的一行文本内容 textFile()方法支持针对目录、压缩文件以及通配符创建RDD Spark默认会为HDFS文件的每一个Block创建一个partition,也可以通过textFile()的第二个参数手动设置分区数量,只能比Block数量多,不能比Block数量少,比Block数量少的话设置是不生效的 **scala版本** import org.apache.spark.{ SparkConf, SparkContext} /** * 需求:通过文件创建RDD * * 1:本地文件 * 2:HDFS文件 */ object CreateRddByFileScala { def main(args: Array[String]): Unit = { // 创建SparkContext val conf = new SparkConf() conf.setAppName("CreateRddByFileScala") // 设置任务名称 .setMaster("local") // local表示在本地执行 val sc = new SparkContext(conf) var path = "~/hello.txt" path = "hdfs://bigdata01:9000/test/hello.txt" // 读取文件数据,可以在textFile中指定生成的RDD的分区数量 val rdd = sc.textFile(path,2) // 获取每一行数据的长度,计算文件内数据的总长度 val length = rdd.map(_.length).reduce(_ + _) println(length) sc.stop() } } **Java版本** import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import org.apache.spark.api.java.function.Function2; public class CreateRddByFileJava { public static void main(String[] args) { // 创建SparkContext SparkConf conf = new SparkConf(); conf.setAppName("CreateRddByArrayJava") .setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); String path = "~/hello.txt"; path = "hdfs://bigdata01:9000/test/hello.txt"; // 读取文件数据,可以在textFile中指定生成的RDD的分区数量 JavaRDD<String> rdd = sc.textFile(path,2); // 获取每一行数据的长度 JavaRDD<Integer> lengthRDD = rdd.map(new Function<String, Integer>() @Override public Integer call(String line) throws Exception { return line.length(); } }); // 计算文件内数据的总长度 Integer length = lengthRDD.reduce(new Function2<Integer, Integer, Int @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }); System.out.println(length); sc.stop(); } }
还没有评论,来说两句吧...