我是靠谱客的博主 自然乌龟,最近开发中收集的这篇文章主要介绍5、创建RDD(集合、本地文件、HDFS文件),觉得挺不错的,现在分享给大家,希望可以做个参考。

概述

一、创建RDD

1、创建RDD

进行Spark核心编程时,首先要做的第一件事,就是创建一个初始的RDD。该RDD中,通常就代表和包含了Spark应用程序的输入源数据。然后在创建了初始的RDD之后,才可以通过Spark Core提供的transformation算子,对该RDD进行转换,来获取其他的RDD。
Spark Core提供了三种创建RDD的方式,包括:使用程序中的集合创建RDD;使用本地文件创建RDD;使用HDFS文件创建RDD。
1、使用程序中的集合创建RDD,主要用于进行测试,可以在实际部署到集群运行之前,自己使用集合构造测试数据,来测试后面的spark应用的流程。
2、使用本地文件创RDD,主要用于临时性地处理一些存储了大量数据的文件。
3、使用HDFS文件创建RDD,应该是最常用的生产环境处理方式,主要可以针对HDFS上存储的大数据,进行离线批处理操作。


2、并行化集合创建RDD

如果要通过并行化集合来创建RDD,需要针对程序中的集合,调用SparkContext的parallelize()方法。Spark会将集合中的数据拷贝到集群上去,形成一个分布式的数据集合,也就是一个RDD。相当于是,集合中的部分数据会到一个节点上,而另一部分数据会到其他节点上。然后就可以用并行的方式来操作这个分布式数据集合,即RDD。
调用parallelize()时,有一个重要的参数可以指定,就是要将集合切分成多少个partition。Spark会为每一个partition运行一个task来进行处理。Spark官方的建议是,为集群中
的每个CPU创建2~4个partition。Spark默认会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时,传入第二个参数,来设置RDD的partition数量。
比如parallelize(arr, 10)
// 案例:1到10累加求和

###java实现
package cn.spark.study.core;
import java.util.Arrays;
import java.util.List;
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;
/**
* 并行化集合创建RDD
*
* @author bcqf
*
*/
public class ParallelizeCollection {
public static void main(String[] args) {
// 创建SparkConf
SparkConf conf = new SparkConf().setAppName("ParallelizeCollection").setMaster("local");
// 创建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 要通过并行化集合的方式创建RDD,那么就调用SparkContext以及其子类,的parallelize()方法
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
JavaRDD<Integer> numberRDD = sc.parallelize(numbers);
// 执行reduce算子操作
// 相当于,先进行1 + 2 = 3;然后再用3 + 3 = 6;然后再用6 + 4 = 10...以此类推;
int sum = numberRDD.reduce(new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer num1, Integer num2) throws Exception {
// TODO Auto-generated method stub
return num1 + num2;
}
});
//输入累加和
System.out.println("1-10的累加和:" + sum);
// 关闭JavaSparkContext

sc.close();
}
}
###scala实现
package cn.spark.study.core
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object ParallelizeCollection {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("ParallelizeCollection").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, 5)
val sum = numberRDD.reduce(_ + _)
println("1-10的累加和:" + sum)
}
}


3、使用本地文件和HDFS创建RDD

Spark是支持使用任何Hadoop支持的存储系统上的文件创建RDD的,比如说HDFS、Cassandra、HBase以及本地文件。通过调用SparkContext的textFile()方法,可以针对本地文件或HDFS文件创建RDD。
有几个事项是需要注意的:
1、如果是针对本地文件的话,如果是在windows上本地测试,windows上有一份文件即可;如果是在spark集群上针对linux本地文件,那么需要将文件拷贝到所有worker节点上。
2、Spark的textFile()方法支持针对目录、压缩文件以及通配符进行RDD创建。
3、Spark默认会为hdfs文件的每一个block创建一个partition,但是也可以通过textFile()的第二个参数手动设置分区数量,只能比block数量多,不能比block数量少。
-------------------案例:文件字数统计
;本地文件-------------------
##java实现
package cn.spark.study.core;
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;
/**
* 使用本地文件创建RDD
* 案例:统计文本文件字数
* @author bcqf
*
*/
public class LocalFile {
public static void main(String[] args) {
// 创建SparkConf
SparkConf conf = new SparkConf().setAppName("LocalFile").setMaster("local");
// 创建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 调用SparkContext以及其子类的textFile()方法,针对本地文件创建RDD
JavaRDD<String> lines = sc.textFile("D://spark.txt");
// 统计文本文件内的字数
JavaRDD<Integer> lineLength = lines.map(new Function<String, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(String v1) throws Exception {
return v1.length();
}
});
int count = lineLength.reduce(new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
System.out.println("文字总数是:" + count);
// 关闭JavaSparkContext

sc.close();
}
}
-------------------案例:文件字数统计
;HDFS文件-------------------
##java实现
package cn.spark.study.core;
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;
/**
* 使用HDFS文件创建RDD
* 案例:统计文本文件字数
* @author bcqf
*
*/
public class HDFSFile {
public static void main(String[] args) {
// 创建SparkConf
SparkConf conf = new SparkConf().setAppName("HDFSFile");
// 创建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 调用SparkContext以及其子类的textFile()方法,针对HDFS文件创建RDD
JavaRDD<String> lines = sc.textFile("hdfs://spark1:9000/spark.txt");
// 统计文本文件内的字数
JavaRDD<Integer> lineLength = lines.map(new Function<String, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(String v1) throws Exception {
return v1.length();
}
});
int count = lineLength.reduce(new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
System.out.println("文字总数是:" + count);
// 关闭JavaSparkContext

sc.close();
}
}
[root@spark1 java]# cat hdfs_file.sh
#运行脚本
/usr/local/spark/bin/spark-submit 
--class cn.spark.study.core.HDFSFile 
--num-executors 3 
--driver-memory 100m 
--executor-memory 100m 
--executor-cores 3 
/usr/local/spark-study/java/saprk-study-java-0.0.1-SNAPSHOT-jar-with-dependencies.jar 
##打maven包-->上传-->运行
-------------------案例:文件字数统计
;本地文件-------------------
##scala实现
package cn.spark.study.core
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object LocalFile {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("LocalFile").setMaster("local")
val sc = new SparkContext(conf)
val lines = sc.textFile("D://spark.txt", 1)
val count = lines.map {line => line.length() }.reduce(_ + _)
println("file's count is: " + count)
}
}
-------------------案例:文件字数统计
;HDFS文件-------------------
##scala实现
package cn.spark.study.core
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object HDFSFile {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("HDFSFile")
val sc = new SparkContext(conf)
val lines = sc.textFile("hdfs://spark1:9000/spark.txt", 1)
val count = lines.map {line => line.length() }.reduce(_ + _)
println("file's count is: " + count)
}
}
[root@spark1 scala]# cat hdfs_file.sh
#运行脚本
/usr/local/spark/bin/spark-submit 
--class cn.spark.study.core.HDFSFile 
--num-executors 3 
--driver-memory 100m 
--executor-memory 100m 
--executor-cores 3 
/usr/local/spark-study/scala/spark-study-scala.jar 
##打jar包(spark-study-scala.jar)-->上传-->运行

转载于:https://www.cnblogs.com/weiyiming007/p/11097403.html

最后

以上就是自然乌龟为你收集整理的5、创建RDD(集合、本地文件、HDFS文件)的全部内容,希望文章能够帮你解决5、创建RDD(集合、本地文件、HDFS文件)所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错,欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供,作为学习参考使用,或来自网络收集整理,版权属于原作者所有。
点赞(51)

评论列表共有 0 条评论

立即
投稿
返回
顶部