Spark RDD使用详解2--RDD创建方式
RDD創建方式
1）從Hadoop文件系統（如HDFS、Hive、HBase）輸入創建。
2）從父RDD轉換得到新RDD。
3）通過parallelize或makeRDD將單機數據創建為分布式RDD。
4）基于DB(Mysql)、NoSQL(HBase)、S3(SC3)、數據流創建。
從集合創建RDD
- parallelize
def parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism)(implicit arg0: ClassTag[T]): RDD[T]
從一個Seq集合創建RDD。
參數1：Seq集合，必須。
參數2：分區數，默認為該Application分配到的資源的CPU核數
?
?
scala> var rdd = sc.parallelize(1 to 10)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at :21
scala> rdd.collect
res3: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
scala> rdd.partitions.size
res4: Int = 15
//設置RDD為3個分區
scala> var rdd2 = sc.parallelize(1 to 10,3)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at :21
scala> rdd2.collect
res5: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
scala> rdd2.partitions.size
res6: Int = 3
?
- makeRDD
def makeRDD[T](seq: Seq[T], numSlices: Int = defaultParallelism)(implicit arg0: ClassTag[T]): RDD[T]
這種用法和parallelize完全相同
def makeRDD[T](seq: Seq[(T, Seq[String])])(implicit arg0: ClassTag[T]): RDD[T]
該用法可以指定每一個分區的preferredLocations。
?
scala> var collect = Seq((1 to 10,Seq("slave007.lxw1234.com","slave002.lxw1234.com")),
(11 to 15,Seq("slave013.lxw1234.com","slave015.lxw1234.com")))
collect: Seq[(scala.collection.immutable.Range.Inclusive, Seq[String])] = List((Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10),
List(slave007.lxw1234.com, slave002.lxw1234.com)), (Range(11, 12, 13, 14, 15),List(slave013.lxw1234.com, slave015.lxw1234.com)))
scala> var rdd = sc.makeRDD(collect)
rdd: org.apache.spark.rdd.RDD[scala.collection.immutable.Range.Inclusive] = ParallelCollectionRDD[6] at makeRDD at :23
scala> rdd.partitions.size
res33: Int = 2
scala> rdd.preferredLocations(rdd.partitions(0))
res34: Seq[String] = List(slave007.lxw1234.com, slave002.lxw1234.com)
scala> rdd.preferredLocations(rdd.partitions(1))
res35: Seq[String] = List(slave013.lxw1234.com, slave015.lxw1234.com)
指定分區的優先位置，對后續的調度優化有幫助。
?
?
從外部存儲創建RDD
- textFile
?
//從hdfs文件創建.
//從hdfs文件創建
scala> var rdd = sc.textFile("hdfs:///tmp/lxw1234/1.txt")
rdd: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[26] at textFile at :21
scala> rdd.count
res48: Long = 4
//從本地文件創建
scala> var rdd = sc.textFile("file:///etc/hadoop/conf/core-site.xml")
rdd: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[28] at textFile at :21
scala> rdd.count
res49: Long = 97
注意這里的本地文件路徑需要在Driver和Executor端存在。
- 從其他HDFS文件格式創建
hadoopFile
sequenceFile
objectFile
newAPIHadoopFile
- 從Hadoop接口API創建
hadoopRDD
newAPIHadoopRDD
比如：從HBase創建RDD
scala> import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
scala> import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
scala> import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.client.HBaseAdmin
scala> val conf = HBaseConfiguration.create()
scala> conf.set(TableInputFormat.INPUT_TABLE,"lxw1234")
scala> var hbaseRDD = sc.newAPIHadoopRDD(
conf,classOf[org.apache.hadoop.hbase.mapreduce.TableInputFormat],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],classOf[org.apache.hadoop.hbase.client.Result])
scala> hbaseRDD.count
res52: Long = 1
?
創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎總結
以上是生活随笔為你收集整理的Spark RDD使用详解2--RDD创建方式的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： spark RDD创建方式：parall
- 下一篇： Spark常用函数讲解之键值RDD转换