這篇文章主要介紹“Spark RDD的創(chuàng)建方式及算子的使用方法是什么”，在日常操作中，相信很多人在Spark RDD的創(chuàng)建方式及算子的使用方法是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對(duì)大家解答”Spark RDD的創(chuàng)建方式及算子的使用方法是什么”的疑惑有所幫助！接下來，請(qǐng)跟著小編一起來學(xué)習(xí)吧！
寧縣網(wǎng)站建設(shè)公司成都創(chuàng)新互聯(lián),寧縣網(wǎng)站設(shè)計(jì)制作，有大型網(wǎng)站制作公司豐富經(jīng)驗(yàn)。已為寧縣上1000+提供企業(yè)網(wǎng)站建設(shè)服務(wù)。企業(yè)網(wǎng)站搭建\外貿(mào)網(wǎng)站制作要多少錢，請(qǐng)找那個(gè)售后服務(wù)好的寧縣做網(wǎng)站的公司定做！
一：簡單了解RDD和RDD處理數(shù)據(jù)
RDD，全稱為Resilient Distributed Datasets，是一個(gè)容錯(cuò)的、并行的數(shù)據(jù)結(jié)構(gòu)，可以讓用戶顯式地將數(shù)據(jù)存儲(chǔ)到磁盤和內(nèi)存中，并能控制數(shù)據(jù)的分區(qū)。
RDD：Spark的核心概念是RDD (resilientdistributed dataset)，指的是一個(gè)只讀的，可分區(qū)的分布式數(shù)據(jù)集，這個(gè)數(shù)據(jù)集的全部或部分可以緩存在內(nèi)存中，在多次計(jì)算間重用。
RDD本質(zhì)上是一個(gè)內(nèi)存數(shù)據(jù)集，在訪問RDD時(shí)，指針只會(huì)指向與操作相關(guān)的部分。例如存在一個(gè)面向列的數(shù)據(jù)結(jié)構(gòu)，其中一個(gè)實(shí)現(xiàn)為Int的數(shù)組，另一個(gè)實(shí)現(xiàn)為Float的數(shù)組。如果只需要訪問Int字段，RDD的指針可以只訪問Int數(shù)組，避免了對(duì)整個(gè)數(shù)據(jù)結(jié)構(gòu)的掃描。
RDD將操作分為兩類：transformation與action。無論執(zhí)行了多少次transformation操作，RDD都不會(huì)真正執(zhí)行運(yùn)算，只有當(dāng)action操作被執(zhí)行時(shí)，運(yùn)算才會(huì)觸發(fā)。而在RDD的內(nèi)部實(shí)現(xiàn)機(jī)制中，底層接口則是基于迭代器的，從而使得數(shù)據(jù)訪問變得更高效，也避免了大量中間結(jié)果對(duì)內(nèi)存的消耗。
在實(shí)現(xiàn)時(shí)，RDD針對(duì)transformation操作，都提供了對(duì)應(yīng)的繼承自RDD的類型，例如map操作會(huì)返回MappedRDD，而flatMap則返回FlatMappedRDD。當(dāng)我們執(zhí)行map或flatMap操作時(shí)，不過是將當(dāng)前RDD對(duì)象傳遞給對(duì)應(yīng)的RDD對(duì)象而已。
注意：創(chuàng)建的Maven工程，以下是 pom.xml 中的依賴：
junit junit 4.12 org.apache.spark spark-core_2.10 1.6.1 org.apache.hadoop hadoop-client 2.6.4 org.apache.spark spark-sql_2.10 1.6.1
二：從Hadoop文件系統(tǒng)（或與Hadoop兼容的其他持久化存儲(chǔ)系統(tǒng)，如Hive，HBase）輸出（HDFS）創(chuàng)建。
eg： 求HDFS文件中內(nèi)容所有行數(shù)據(jù)長度及總長度。
重慶分公司
重慶分公司
public class TestRDD1 {
public static void main(String[] args) {
createRDDFromHDFS();
}
private static void createRDDFromHDFS(){
SparkConf conf = new SparkConf();
conf.set("spark.testing.memory", "269522560000");
JavaSparkContext sc = new JavaSparkContext("local", "Spark Test", conf);
System.out.println( sc );
JavaRDD rdd = sc.textFile("hdfs://192.168.226.129:9000/txt/sparkshell/sparkshell.txt");
JavaRDD newRDD = rdd.map( new Function(){
private static final long serialVersionUID = 1L;
public Integer call(String string) throws Exception {
System.out.println( string + " " + string.length() );
return string.length();
}
});
System.out.println( newRDD.count() );
int length = newRDD.reduce( new Function2(){
private static final long serialVersionUID = 1L;
public Integer call(Integer int1, Integer int2) throws Exception {
return int1+int2;
}
});
System.out.println("總和" + length);
}
} 三：通過parallelize或makeRDD將單擊數(shù)據(jù)創(chuàng)建為分布式RDD。
eg：求總和。
public class TestRDD2 {
public static void main(String[] args) {
createRDDFromSuperRDD();
}
/**
* JavaSparkContext(String master, String appName, SparkConf conf)
* master - Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4]).
* appName - A name for your application, to display on the cluster web UI
* conf - a SparkConf object specifying other Spark parameters
* */
private static void createRDDFromSuperRDD(){
SparkConf conf = new SparkConf();
conf.set("spark.testing.memory", "269522560000");
JavaSparkContext sc = new JavaSparkContext("local", "Spark Test", conf);
System.out.println( sc );
List list = new ArrayList();
for( int i=1;i<=10;i++){
list.add(i);
}
JavaRDD rdd = sc.parallelize(list);
JavaRDD newRDD = rdd.map( new Function(){
private static final long serialVersionUID = 1L;
public Integer call(Integer int1) throws Exception {
return int1;
}
});
int count = newRDD.reduce( new Function2(){
private static final long serialVersionUID = 1L;
public Integer call(Integer int1, Integer int2) throws Exception {
return int1+int2;
}
});
System.out.println("總和" + count);
}
} 注意： 上述兩段代碼中，在獲取 JavaSparkContext的時(shí)候，是這樣寫的：
SparkConf conf = new SparkConf();
conf.set("spark.testing.memory", "269522560000"); // 給jvm足夠的資源。
JavaSparkContext sc = new JavaSparkContext("local", "Spark Test", conf);
而對(duì)于標(biāo)記的加粗紅色部分，參照API如下：
JavaSparkContext(String master, String appName, SparkConf conf)
-master - Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4]).
-appName - A name for your application, to display on the cluster web UI
-conf - a SparkConf object specifying other Spark parameters
對(duì)于master，官網(wǎng)有詳細(xì)的介紹：
我這里寫的是 "local"，表示的是：
對(duì)于本地模式測試和單元測試，可以通過"local"在spark內(nèi)運(yùn)行程序。
******************************
另外寫的一段，對(duì)算子中一些基本方法的使用
參考學(xué)習(xí)：
RDD算子分類： http://my.oschina.net/gently/blog/686800 （自己的。）
public class TestRDD3 {
private static String appName = "Test Spark RDD";
private static String master = "local";
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.set("spark.testing.memory", "269522560000");
JavaSparkContext sc = new JavaSparkContext(master, appName, conf);
System.out.println( sc );
List list = new ArrayList();
list.add( "Berg" );
list.add( "Hadoop" );
list.add( "HBase" );
list.add( "Hive" );
list.add( "Spark" );
JavaRDD rdd = sc.parallelize(list);
JavaRDD newrdd = rdd.map( new Function(){
private static final long serialVersionUID = 1L;
public Integer call(String string) throws Exception {
System.out.println( string + "\t" +string.length() );
return string.length();
}
});
Integer length = newrdd.reduce( new Function2() {
private static final long serialVersionUID = 1L;
public Integer call(Integer i1, Integer i2) throws Exception {
return i1+i2;
}
});
long count = newrdd.count();
List listnewrdd = newrdd.collect();
for (Integer integer : listnewrdd) {
System.out.print(integer + " \t" );
}
System.out.println( "\nlength --> " + length + " " + count );
System.out.println( "\n\n**************************************\n\n");
List list1 = new ArrayList();
for( int i=1; i<=5;i++){
list1.add( i );
}
JavaRDD rdd1 = sc.parallelize(list1);
JavaRDD unionrdd = newrdd.union(rdd1);
JavaRDD rdd2 = unionrdd.map( new Function(){
private static final long serialVersionUID = 1L;
public Integer call(Integer i) throws Exception {
return i;
}
});
long count2 = rdd2.reduce( new Function2() {
private static final long serialVersionUID = 1L;
public Integer call(Integer arg0, Integer arg1) throws Exception {
return arg0 + arg1;
}
});
System.out.println("count2 --> " +count2 );
rdd2.foreach( new VoidFunction(){
private static final long serialVersionUID = 1L;
public void call(Integer arg0) throws Exception {
System.out.println( "foreach--> " + arg0 );
}
});
}
} 到此，關(guān)于“Spark RDD的創(chuàng)建方式及算子的使用方法是什么”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！
