Spark RDD的創建方式及算子的使用方法是什么

這篇文章主要介紹“Spark RDD的創建方式及算子的使用方法是什么”，在日常操作中，相信很多人在Spark RDD的創建方式及算子的使用方法是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Spark RDD的創建方式及算子的使用方法是什么”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

一：簡單了解RDD和RDD處理數據

    RDD，全稱為Resilient Distributed Datasets，是一個容錯的、并行的數據結構，可以讓用戶顯式地將數據存儲到磁盤和內存中，并能控制數據的分區。

    RDD：Spark的核心概念是RDD (resilientdistributed dataset)，指的是一個只讀的，可分區的分布式數據集，這個數據集的全部或部分可以緩存在內存中，在多次計算間重用。

    RDD本質上是一個內存數據集，在訪問RDD時，指針只會指向與操作相關的部分。例如存在一個面向列的數據結構，其中一個實現為Int的數組，另一個實現為Float的數組。如果只需要訪問Int字段，RDD的指針可以只訪問Int數組，避免了對整個數據結構的掃描。

    RDD將操作分為兩類：transformation與action。無論執行了多少次transformation操作，RDD都不會真正執行運算，只有當action操作被執行時，運算才會觸發。而在RDD的內部實現機制中，底層接口則是基于迭代器的，從而使得數據訪問變得更高效，也避免了大量中間結果對內存的消耗。

    在實現時，RDD針對transformation操作，都提供了對應的繼承自RDD的類型，例如map操作會返回MappedRDD，而flatMap則返回FlatMappedRDD。當我們執行map或flatMap操作時，不過是將當前RDD對象傳遞給對應的RDD對象而已。

注意：創建的Maven工程，以下是 pom.xml 中的依賴：

<dependencies>

		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>4.12</version>
		</dependency>

		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-core_2.10</artifactId>
			<version>1.6.1</version>

		</dependency>

		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>2.6.4</version>
		</dependency>

		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-sql_2.10</artifactId>
			<version>1.6.1</version>
		</dependency>

	</dependencies>

二：從Hadoop文件系統（或與Hadoop兼容的其他持久化存儲系統，如Hive，HBase）輸出（HDFS）創建。

    eg： 求HDFS文件中內容所有行數據長度及總長度。

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<String> rdd = sc.textFile("hdfs://192.168.226.129:9000/txt/sparkshell/sparkshell.txt");
		
		JavaRDD<Integer> newRDD = rdd.map( new Function<String,Integer>(){

			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<Integer, Integer, Integer>(){

			private static final long serialVersionUID = 1L;

			public Integer call(Integer int1, Integer int2) throws Exception {
				return int1+int2;
			}
			
		});
		
		System.out.println("總和" + length);
	}
	
}

三：通過parallelize或makeRDD將單擊數據創建為分布式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<Integer> list = new ArrayList<Integer>();
		
		for( int i=1;i<=10;i++){
			list.add(i);
		}
		
		JavaRDD<Integer> rdd = sc.parallelize(list);
		
		JavaRDD<Integer> newRDD = rdd.map( new Function<Integer,Integer>(){

			private static final long serialVersionUID = 1L;

			public Integer call(Integer int1) throws Exception {
				return int1;
			}
		});
		
		int count = newRDD.reduce( new Function2<Integer, Integer, Integer>(){

			private static final long serialVersionUID = 1L;

			public Integer call(Integer int1, Integer int2) throws Exception {
				return int1+int2;
			}
			
		});
		
		System.out.println("總和" + count);
	}
	
}

注意： 上述兩段代碼中，在獲取 JavaSparkContext的時候，是這樣寫的：

        SparkConf conf = new SparkConf();

        conf.set("spark.testing.memory", "269522560000");  // 給jvm足夠的資源。

        JavaSparkContext sc = new JavaSparkContext("local", "Spark Test", conf);

而對于標記的加粗紅色部分，參照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

對于master，官網有詳細的介紹：

我這里寫的是 "local"，表示的是：

    對于本地模式測試和單元測試，可以通過"local"在spark內運行程序。

******************************

另外寫的一段，對算子中一些基本方法的使用

參考學習：

    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<String> list = new ArrayList<String>();
		list.add( "Berg" );
		list.add( "Hadoop" );
		list.add( "HBase" );
		list.add( "Hive" );
		list.add( "Spark" );
		
		JavaRDD<String> rdd =  sc.parallelize(list);
		
		JavaRDD<Integer> newrdd = rdd.map( new Function<String,Integer>(){
			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<Integer, Integer, Integer>() {

			private static final long serialVersionUID = 1L;
			public Integer call(Integer i1, Integer i2) throws Exception {
				return i1+i2;
			}
		});
		
		long count = newrdd.count();
		List<Integer> 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<Integer> list1 = new ArrayList<Integer>();
		for( int i=1; i<=5;i++){
			list1.add( i );
		}
		
		JavaRDD<Integer> rdd1 = sc.parallelize(list1);
		JavaRDD<Integer> unionrdd = newrdd.union(rdd1);
		
		JavaRDD<Integer> rdd2 = unionrdd.map( new Function<Integer,Integer>(){

			private static final long serialVersionUID = 1L;
			public Integer call(Integer i) throws Exception {
				return i;
			}
		});
		
		long count2 = rdd2.reduce( new Function2<Integer, Integer, Integer>() {

			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<Integer>(){
			private static final long serialVersionUID = 1L;
			public void call(Integer arg0) throws Exception {
				System.out.println(  "foreach--> " + arg0 );
			}
			
		});
	}
}

到此，關于“Spark RDD的創建方式及算子的使用方法是什么”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！