Spark重点难点知识总结（二）

本文主要是介绍Spark重点难点知识总结（二），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1.join：join函数主要用来拼接字符串，将字符串、元组、列表中的元素以指定的字符(分隔符)连接生成一个新的字符串。

var rdd1 = sc.makeRDD(Array(("A","1"),("B","2"),("C","3")),2)
var rdd2 = sc.makeRDD(Array(("A","a"),("C","c"),("D","d")),2)
scala> rdd1.join(rdd2).collect
res: Array[(String, (String, String))] = Array((A,(1,a)), (C,(3,c)))

2.countByKey：统计每个key对应的value个数。

val scores=Array(Tuple2(1,100),Tuple2(2,90),Tuple2(3,100),Tuple2(2,90),Tuple2(3,100))
val content=sc.parallelize(scores)
val data=content.countByKey()
res：data: scala.collection.Map[Int,Long] = Map(1 -> 1, 3 -> 2, 2 -> 2)

3.宽依赖和窄依赖：窄依赖是指每个父RDD的一个partition最多被子RDD的一个partition所使用，例如map，filter，union等都会产生窄依赖。窄依赖，1对1，n对1。宽依赖是指一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、ReduceByKey、sortByKey等操作都会产生宽依赖。总结：如果父RDD的一个Partition被一个子RDD的Partition所使用就是窄依赖，否则的话就是宽依赖。

4.DataFrame与RDD：DataFrame是一种分布式二维数据结构，R和Python语言中都有DataFrame，Spark中的DataFrame最大的不同点是其天生是分布式的，可以简单的认为Spark中的DataFrame是一个分布式的Table,形式如下所示。

而RDD类型为

简单来说，RDD是一个个Person实例，RDD并不知道里面有什么类型的数据。

（1）RDD以Record为单位，Spark在优化的时候无法洞悉Record内部的细节，所以也就无法进行更深度的优化，这极大的限制了Spark SQL性能的提示。
（2）DataFrame包含了每个Record的Metadata信息，也就是说DataFrame的优化是基于列内部的优化，而不是像RDD一样，只能够基于行进行优化。

5.RDD转DataFrame

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.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.Row;
/** 使用反射的方式将RDD转化为DataFrame*/
public class RDD2DataFrame {public static void main(String[] args) {SparkConf conf=new SparkConf().setAppName("RDD2DataFrame").setMaster("local");JavaSparkContext sc=new JavaSparkContext(conf);SQLContext sqlContext=new SQLContext(sc);JavaRDD<String> lines=sc.textFile("C://Users//Jason Shu//Desktop//persons.txt");JavaRDD<Person> persons=lines.map(new Function<String, Person>(){public Person call(String line) throws Exception {String[] splited=line.split(",");Person p =new Person();p.setId(Integer.valueOf(splited[0].trim()));p.setName(splited[1]);p.setAge(Integer.valueOf(splited[0].trim()));return p;}});DataFrame df= sqlContext.createDataFrame(persons, Person.class);df.registerTempTable("persons");//注册一张临时表DataFrame bigData=sqlContext.sql("select * from persons where age >=6");JavaRDD<Row> bigDataRDD=bigData.javaRDD();JavaRDD<Person> result=bigDataRDD.map(new Function<Row, Person>() {public Person call(Row row) throws Exception {Person p =new Person();p.setId(row.getInt(0));p.setName(row.getString(1));p.setAge(row.getInt(2));return p;}});List<Person> personList=result.collect();for(Person p:personList){System.out.println(p);}   }}

    转换过程示意图

6.Spark SQL和DataFrame：Spark SQL 是 Spark 生态系统里用于处理结构化大数据的模块，该模块里最重要的概念就是DataFrame, Spark 的 DataFrame 是基于早期版本中的 SchemaRDD，所以很自然的使用分布式大数据处理的场景。DataFrame 以 RDD 为基础，但是带有 Schema 信息，它类似于传统数据库中的二维表格。Spark SQL 模块目前支持将多种外部数据源的数据转化为 DataFrame，并像操作 RDD 或者将其注册为临时表的方式处理和分析这些数据。当前支持的数据源有：Json，文本文件，RDD，关系数据库，Hive，Parquet。一旦将 DataFrame 注册成临时表，我们就可以使用类 SQL 的方式操作这些数据。Spark SQL的表数据在内存中存储不是采用原生态的JVM对象存储方式，而是采用内存列存储，如下图所示。

7.Schema的方式创建DataFrame

import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkContext, SparkConf}object schemaDataFrame {def main(args:Array[String]): Unit ={val conf=new SparkConf()conf.setMaster("local").setAppName("schemaDataFrame ")val sc=new SparkContext(conf)val RowRDD =sc.textFile("C://Users//Jason Shu//Desktop//InputFile.txt").map(x=>x.split(" ")).map(p=>Row(p(0),p(1)))val sqlContext=new SQLContext(sc)val peopleSchema=StructType(Array(StructField("name", StringType, true),StructField("age", IntegerType, true),StructField("sex", BooleanType, true)))val peopleDataFrame = sqlContext.createDataFrame(RowRDD, peopleSchema)//创建DataFrame,第一个参数为Row[RDD],第二个参数为StructTypepeopleDataFrame.registerTempTable("people")//表的名字随便取一个val results = sqlContext.sql("SELECT name FROM people")results.map(t => "name: " + t(0)).collect().foreach(println)}
}

8.first():Return the first element in this RDD，first返回RDD中的第一个元素，不排序。

scala> var rdd1 = sc.makeRDD(Array(("A","1"),("B","2"),("C","3")),2)
rdd1: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[33] at makeRDD at :21scala> rdd1.first
res14: (String, String) = (A,1)scala> var rdd1 = sc.makeRDD(Seq(10, 4, 2, 12, 3))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at :21scala> rdd1.first
res8: Int = 10

9.contains():Returns true if and only if this string contains the specified sequence of char values，当且仅当此字符串包含指定的char值序列返回true

10.parallellize():在一个已经存在的Scala集合上创建的RDD， 集合的对象将会被拷贝，创建出一个可以被并行操作的分布式数据集。

val data = List(1, 2, 3, 4, 5)  val distData = sc.parallelize(data)//distData此时为RDD[Int]

11.socketTextStream():相当于Socket客户端，里面的参数就是socket服务器的ip和端口，执行该语句的时候就向socket服务器发送了建立请求了。服务器端接受到了请求就可以给socketTextStream发送消息了

12.filter():使用filter方法，你可以筛选出集合中你需要的元素，形成一个新的集合。

val x = List.range(1, 10)val evens = x.filterNot(_ % 2 == 0)Res：evens: List[Int] = List(1, 3, 5, 7, 9)

13.String.valueOf():要把参数中给的值，转化为String类型，这里的参数是Any，任意的参数都可以。

14.Integer.parseInt：将整数的字符串，转化为整数。

val b="123"val a=Integer.parseInt(b)println(a)//打印结果123

15.collect():将RDD转成Scala数组，并返回。

16.Spark中的partition：提供一种划分数据的依据。例如wordcount程序中的：

val lines=sc.textFile(path, 8)

这个地方的8就是指8个分区，当然如果数据量不够或不够复杂，可以不分为8个。

这篇关于Spark重点难点知识总结（二）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---