1.SparkShuffle
1.1SparkShuffle概念：
reduceByKey会将上一个RDD中的每一个key对应的所有value聚合成一个value，然后生成一个新的RDD，元素类型是<key,value>对的形式，这样每一个key对应一个聚合起来的value。
问题：
聚合之前，每一个key对应的value不一定都是在一个partition中，也不太可能在同一个节点上，因为RDD是分布式的弹性的数据集，RDD的partition极有可能分布在各个节点上。
如何聚合？
Shuffle Write：上一个stage的每个map task就必须保证将自己处理的当前分区的数据相同的key写入一个分区文件中，可能会写入多个不同的分区文件中。
Shuffle Read：reduce task就会从上一个stage的所有task所在的机器上寻找属于己的那些分区文件，这样就可以保证每一个key所对应的value都会汇聚到同一个节点上去处理和聚合。
Spark中有两种Shuffle管理类型，HashShufflManager和SortShuffleManager，Spark1.2之前是HashShuffleManager， Spark1.2引入SortShuffleManager,在Spark 2.0+版本中已经将HashShuffleManager丢弃。
1.2HashShuffleManager
1) 普通机制
普通机制示意图：
执行流程：
a) 每一个map task将不同结果写到不同的buffer中，每个buffer的大小为32K。buffer起到数据缓存的作用。
b) 每个buffer文件最后对应一个磁盘小文件。
c) reduce task来拉取对应的磁盘小文件。
总结：
map task的计算结果会根据分区器（默认是hashPartitioner）来决定写入到哪一个磁盘小文件中去。ReduceTask会去Map端拉取相应的磁盘小文件。
产生的磁盘小文件的个数：
M（map task的个数）*R（reduce task的个数）
存在的问题：
产生的磁盘小文件过多，会导致以下问题：
在Shuffle Write过程中会产生很多写磁盘小文件的对象。
在Shuffle Read过程中会产生很多读取磁盘小文件的对象。
在JVM堆内存中对象过多会造成频繁的gc,gc还无法解决运行所需要的内存 的话，就会OOM。
在数据传输过程中会有频繁的网络通信，频繁的网络通信出现通信故障的可能性大大增加，一旦网络通信出现了故障会导致shuffle file cannot find 由于这个错误导致的task失败，TaskScheduler不负责重试，由DAGScheduler负责重试Stage。
合并机制：
合并机制示意图：
总结：
产生磁盘小文件的个数：C(core的个数)*R（reduce的个数）
SortShuffleManager
普通机制：
普通机制示意图
执行流程：
map task 的计算结果会写入到一个内存数据结构里面，内存数据结构默认是5M。
在shuffle的时候会有一个定时器，不定期的去估算这个内存结构的大小，当内存结构中的数据超过5M时，比如现在内存结构中的数据为5.01M，那么他会申请5.01*2-5=5.02M内存给内存数据结构。
如果申请成功不会进行溢写，如果申请不成功，这时候会发生溢写磁盘。
在溢写之前内存结构中的数据会进行排序分区。
然后开始溢写磁盘，写磁盘是以batch的形式去写，一个batch是1万条数据，map task执行完成后，会将这些磁盘小文件合并成一个大的磁盘文件，同时生成一个索引文件。
reduce task去map端拉取数据的时候，首先解析索引文件，根据索引文件再去拉取对应的数据。
总结：
产生磁盘小文件的个数： 2*M（map task的个数）
bypass机制;
bypass机制示意图:
总结:
bypass运行机制的触发条件如下：
shuffle reduce task的数量小于spark.shuffle.sort.bypassMergeThreshold的参数值。这个值默认是200。
产生的磁盘小文件为：2*M（map task的个数）
Shuffle文件寻址:
MapOutputTracker:
MapOutputTracker是Spark架构中的一个模块，是一个主从架构。管理磁盘小文件的地址。
MapOutputTrackerMaster是主对象，存在于Driver中。
Ø MapOutputTrackerWorker是从对象，存在于Excutor中。
BlockManager:
BlockManager块管理者，是Spark架构中的一个模块，也是一个主从架构。
Ø BlockManagerMaster,主对象，存在于Driver中。
BlockManagerMaster会在集群中有用到广播变量和缓存数据或者删除缓存数据的时候，通知BlockManagerSlave传输或者删除数据。
Ø BlockManagerSlave，从对象，存在于Excutor中。
BlockManagerSlave会与BlockManagerSlave之间通信。
¬ 无论在Driver端的BlockManager还是在Excutor端的BlockManager都含有三个对象：
①DiskStore:负责磁盘的管理。
②MemoryStore：负责内存的管理。
BlockTransferService:负责数据的传输。
Shuffle文件寻址图:
Shuffle文件寻址流程:
a)当map task执行完成后，会将task的执行情况和磁盘小文件的地址封装到MpStatus对象中，通过MapOutputTrackerWorker对象向Driver中的MapOutputTrackerMaster汇报。
b)在所有的map task执行完毕后，Driver中就掌握了所有的磁盘小文件的地址。
c)在reduce task执行之前，会通过Excutor中MapOutPutTrackerWorker向Driver端的MapOutputTrackerMaster获取磁盘小文件的地址。
d)获取到磁盘小文件的地址后，会通过BlockManager连接数据所在节点，然后通过BlockTransferService进行数据的传输。
e)BlockTransferService默认启动5个task去节点拉取数据。默认情况下，5个task拉取数据量不能超过48M。
Spark内存管理:
Spark执行应用程序时，Spark集群会启动Driver和Executor两种JVM进程，Driver负责创建SparkContext上下文，提交任务，task的分发等。
Executor负责task的计算任务，并将结果返回给Driver。
同时需要为需要持久化的RDD提供储存。
Driver端的内存管理比较简单，这里所说的Spark内存管理针对Executor端的内存管理。
Spark内存管理分为静态内存管理和统一内存管理，Spark1.6之前使用的是静态内存管理，Spark1.6之后引入了统一内存管理。
静态内存管理中存储内存、执行内存和其他内存的大小在 Spark 应用程序运行期间均为固定的，但用户可以应用程序启动前进行配置。
统一内存管理与静态内存管理的区别在于储存内存和执行内存共享同一块空间，可以互相借用对方的空间。
Spark1.6以上版本默认使用的是统一内存管理，可以通过参数, spark.memory.useLegacyMode 设置为true(默认为false)使用静态内存管理。
静态内存管理分布图:
reduce 中OOM如何处理？
1)减少每次拉取的数据量.
2)提高shuffle聚合的内存比例.
3)提高Excutor的总内存.
Shuffle调优:
1.SparkShuffle调优配置项如何使用？
1)在代码中,不推荐使用，硬编码。
new SparkConf().set(“spark.shuffle.file.buffer”,”64”)
2)在提交spark任务的时候，推荐使用。
spark-submit –conf spark.shuffle.file.buffer=64 –conf ….
在conf下的spark-default.conf配置文件中,不推荐，因为是写死后所有应用程序都要用。
shuffle调优：
spark.shuffle.file.buffer
默认值：32k
参数说明：该参数用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小。将数据写到磁盘文件之前，会先写入buffer缓冲中，待缓冲写满之后，才会溢写到磁盘。
调优建议：如果作业可用的内存资源较为充足的话，可以适当增加这个参数的大小（比如64k），从而减少shuffle write过程中溢写磁盘文件的次数，也就可以减少磁盘IO次数，进而提升性能。在实践中发现，合理调节该参数，性能会有1%~5%的提升。
spark.reducer.maxSizeInFlight
默认值：48m
参数说明：该参数用于设置shuffle read task的buffer缓冲大小，而这个buffer缓冲决定了每次能够拉取多少数据。
调优建议：如果作业可用的内存资源较为充足的话，可以适当增加这个参数的大小（比如96m），从而减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能。在实践中发现，合理调节该参数，性能会有1%~5%的提升。
spark.shuffle.io.maxRetries
默认值：3
参数说明：shuffle read task从shuffle write task所在节点拉取属于自己的数据时，如果因为网络异常导致拉取失败，是会自动进行重试的。该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功，就可能会导致作业执行失败。
调优建议：对于那些包含了特别耗时的shuffle操作的作业，建议增加重试最大次数（比如60次），以避免由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现，对于针对超大数据量（数十亿~上百亿）的shuffle过程，调节该参数可以大幅度提升稳定性。
shuffle file not find taskScheduler不负责重试task，由DAGScheduler负责重试stage
spark.shuffle.io.retryWait
默认值：5s
参数说明：具体解释同上，该参数代表了每次重试拉取数据的等待间隔，默认是5s。
调优建议：建议加大间隔时长（比如60s），以增加shuffle操作的稳定性。
spark.shuffle.memoryFraction
默认值：0.2
参数说明：该参数代表了Executor内存中，分配给shuffle read task进行聚合操作的内存比例，默认是20%。
调优建议：在资源参数调优中讲解过这个参数。如果内存充足，而且很少使用持久化操作，建议调高这个比例，给shuffle read的聚合操作更多内存，以避免由于内存不足导致聚合过程中频繁读写磁盘。在实践中发现，合理调节该参数可以将性能提升10%左右。
spark.shuffle.manager
默认值：sort|hash
参数说明：该参数用于设置ShuffleManager的类型。Spark 1.5以后，有三个可选项：hash、sort和tungsten-sort。HashShuffleManager是Spark 1.2以前的默认选项，但是Spark 1.2以及之后的版本默认都是SortShuffleManager了。tungsten-sort与sort类似，但是使用了tungsten计划中的堆外内存管理机制，内存使用效率更高。
调优建议：由于SortShuffleManager默认会对数据进行排序，因此如果你的业务逻辑中需要该排序机制的话，则使用默认的SortShuffleManager就可以；而如果你的业务逻辑不需要对数据进行排序，那么建议参考后面的几个参数调优，通过bypass机制或优化的HashShuffleManager来避免排序操作，同时提供较好的磁盘读写性能。这里要注意的是，tungsten-sort要慎用，因为之前发现了一些相应的bug。
spark.shuffle.sort.bypassMergeThreshold
默认值：200
参数说明：当ShuffleManager为SortShuffleManager时，如果shuffle read task的数量小于这个阈值（默认是200），则shuffle write过程中不会进行排序操作，而是直接按照未经优化的HashShuffleManager的方式去写数据，但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件，并会创建单独的索引文件。
调优建议：当你使用SortShuffleManager时，如果的确不需要排序操作，那么建议将这个参数调大一些，大于shuffle read task的数量。那么此时就会自动启用bypass机制，map-side就不会进行排序了，减少了排序的性能开销。但是这种方式下，依然会产生大量的磁盘文件，因此shuffle write性能有待提高。
spark.shuffle.consolidateFiles
默认值：false
参数说明：如果使用HashShuffleManager，该参数有效。如果设置为true，那么就会开启consolidate机制，会大幅度合并shuffle write的输出文件，对于shuffle read task数量特别多的情况下，这种方法可以极大地减少磁盘IO开销，提升性能。
调优建议：如果的确不需要SortShuffleManager的排序机制，那么除了使用bypass机制，还可以尝试将spark.shffle.manager参数手动指定为hash，使用HashShuffleManager，同时开启consolidate机制。在实践中尝试过，发现其性能比开启了bypass机制的SortShuffleManager要高出10%~30%。
Shark：
Shark是基于Spark计算框架之上且兼容Hive语法的SQL执行引擎，由于底层的计算采用了Spark，性能比MapReduce的Hive普遍快2倍以上，当数据全部load在内存的话，将快10倍以上。
因此Shark可以作为交互式查询应用服务来使用。
除了基于Spark的特性外，Shark是完全兼容Hive的语法，表结构以及UDF函数等，已有的HiveSql可以直接进行迁移至Shark上Shark底层依赖于Hive的解析器，查询优化器，但正是由于SHark的整体设计架构对Hive的依赖性太强，难以支持其长远发展，比如不能和Spark的其他组件进行很好的集成，无法满足Spark的一栈式解决大数据处理的需求。
SparkSQL：
SparkSQL介绍：
Hive是Shark的前身，Shark是SparkSQL的前身,SparkSQL产生的根本原因是其完全脱离了Hive的限制。
SparkSQL支持查询原生的RDD。 RDD是Spark平台的核心概念，是Spark能够高效的处理大数据的各种场景的基础。
能够在Scala中写SQL语句。支持简单的SQL语法检查，能够在Scala中写Hive语句访问Hive数据，并将结果取回作为RDD使用。
Spark on Hive和Hive on Spark：
Spark on Hive： Hive只作为储存角色，Spark负责sql解析优化，执行。
Hive on Spark：Hive即作为存储又负责sql的解析优化，Spark负责执行。
DataFrame：
DataFrame也是一个分布式数据容器。
与RDD类似，然而DataFrame更像传统数据库的二维表格，除了数据以外，还掌握数据的结构信息，即schema。
同时，与Hive类似，DataFrame也支持嵌套数据类型（struct、array和map）。 从API易用性的角度上看， DataFrame API提供的是一套高层的关系操作，比函数式的RDD API要更加友好，门槛更低。
DataFrame就Row类型的DataSet。
SparkSQL的数据源：
SparkSQL的数据源可以是JSON类型的字符串，JDBC,Parquent,Hive，HDFS等。
SparkSQL底层架构：
首先拿到sql后解析一批未被解决的逻辑计划，再经过分析得到分析后的逻辑计划，再经过一批优化规则转换成一批最佳优化的逻辑计划，再经过SparkPlanner的策略转化成一批物理计划，随后经过消费模型转换成一个个的Spark任务执行。
谓词下推（predicate Pushdown）：
创建DataFrame的几种方式：
读取json格式的文件创建DataFrame
注意：
json文件中的json数据不能嵌套json格式数据。
DataFrame是一个一个Row类型的RDD，df.rdd()/df.javaRdd()。
可以两种方式读取json格式的文件。
df.show()默认显示前20行数据。
DataFrame原生API可以操作DataFrame。
注册成临时表时，表中的列默认按ascii顺序显示列。
java:
SparkConf conf = new SparkConf();
conf.setMaster(“local”).setAppName(“jsonfile”);
SparkContext sc = new SparkContext(conf);
//创建sqlContext
SQLContext sqlContext = new SQLContext(sc);
/**
* DataFrame的底层是一个一个的RDD RDD的泛型是Row类型。
* 以下两种方式都可以读取json格式的文件
*/
DataFrame df = sqlContext.read().format(“json”).load(“sparksql/json”);
// DataFrame df2 = sqlContext.read().json(“sparksql/json.txt”);
// df2.show();
/**
* DataFrame转换成RDD
*/
RDD<Row> rdd = df.rdd();
/**
* 显示 DataFrame中的内容，默认显示前20行。如果现实多行要指定多少行show(行数)
* 注意：当有多个列时，显示的列先后顺序是按列的ascii码先后显示。
*/
// df.show();
/**
* 树形的形式显示schema信息
*/
df.printSchema();
/**
* dataFram自带的API 操作DataFrame
*/
//select name from table
// df.select(“name”).show();
//select name age+10 as addage from table
df.select(df.col(“name”),df.col(“age”).plus(10).alias(“addage”)).show();
//select name ,age from table where age>19
df.select(df.col(“name”),df.col(“age”)).where(df.col(“age”).gt(19)).show();
//select count(*) from table group by age
df.groupBy(df.col(“age”)).count().show();
/**
* 将DataFrame注册成临时的一张表，这张表临时注册到内存中，是逻辑上的表，不会雾化到磁盘
*/
df.registerTempTable(“jtable”);
DataFrame sql = sqlContext.sql(“select age,count(1) from jtable group by age”);
DataFrame sql2 = sqlContext.sql(“select * from jtable”);
sc.stop();
scala:
val conf = new SparkConf()
conf.setMaster(“local”).setAppName(“jsonfile”)
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.json(“sparksql/json”)
//val df1 = sqlContext.read.format(“json”).load(“sparksql/json”)
df.show()
df.printSchema()
//select * from table
df.select(df.col(“name”)).show()
//select name from table where age>19
df.select(df.col(“name”),df.col(“age”)).where(df.col(“age”).gt(19)).show()
//select count(*) from table group by age
df.groupBy(df.col(“age”)).count().show();
/**
* 注册临时表
*/
df.registerTempTable(“jtable”)
val result = sqlContext.sql(“select * from jtable”)
result.show()
sc.stop()
通过json格式的RDD创建DataFrame
java:
SparkConf conf = new SparkConf();
conf.setMaster(“local”).setAppName(“jsonRDD”);
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList(
“{\”name\”:\”zhangsan\”,\”age\”:\”18\”}”,
“{\”name\”:\”lisi\”,\”age\”:\”19\”}”,
“{\”name\”:\”wangwu\”,\”age\”:\”20\”}”
));
JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList(
“{\”name\”:\”zhangsan\”,\”score\”:\”100\”}”,
“{\”name\”:\”lisi\”,\”score\”:\”200\”}”,
“{\”name\”:\”wangwu\”,\”score\”:\”300\”}”
));
DataFrame namedf = sqlContext.read().json(nameRDD);
DataFrame scoredf = sqlContext.read().json(scoreRDD);
namedf.registerTempTable(“name”);
scoredf.registerTempTable(“score”);
DataFrame result = sqlContext.sql(“select name.name,name.age,score.score from name,score where name.name = score.name”);
result.show();
sc.stop();
scala:
val conf = new SparkConf()
conf.setMaster(“local”).setAppName(“jsonrdd”)
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val nameRDD = sc.makeRDD(Array(
“{\”name\”:\”zhangsan\”,\”age\”:18}”,
“{\”name\”:\”lisi\”,\”age\”:19}”,
“{\”name\”:\”wangwu\”,\”age\”:20}”
))
val scoreRDD = sc.makeRDD(Array(
“{\”name\”:\”zhangsan\”,\”score\”:100}”,
“{\”name\”:\”lisi\”,\”score\”:200}”,
“{\”name\”:\”wangwu\”,\”score\”:300}”
))
val nameDF = sqlContext.read.json(nameRDD)
val scoreDF = sqlContext.read.json(scoreRDD)
nameDF.registerTempTable(“name”)
scoreDF.registerTempTable(“score”)
val result = sqlContext.sql(“select name.name,name.age,score.score from name,score where name.name = score.name”)
result.show()
sc.stop()
非json格式的RDD创建DataFrame:
通过反射的方式将非json格式的RDD转换成DataFrame（不建议使用）.
自定义类要可序列化。
自定义类的访问级别是Public。
RDD转成DataFrame后会根据映射将字段按Assci码排序。
将DataFrame转换成RDD时获取字段两种方式,一种是df.getInt(0)下标获取（不推荐使用），另一种是df.getAs(“列名”)获取（推荐使用）。
/**
* 注意：
* 1.自定义类必须是可序列化的
* 2.自定义类访问级别必须是Public
* 3.RDD转成DataFrame会把自定义类中字段的名称按assci码排序
*/
SparkConf conf = new SparkConf();
conf.setMaster(“local”).setAppName(“RDD”);
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile(“sparksql/person.txt”);
JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Person call(String s) throws Exception {
Person p = new Person();
p.setId(s.split(“,”)[0]);
p.setName(s.split(“,”)[1]);
p.setAge(Integer.valueOf(s.split(“,”)[2]));
return p;
}
});
/**
* 传入进去Person.class的时候，sqlContext是通过反射的方式创建DataFrame
* 在底层通过反射的方式获得Person的所有field，结合RDD本身，就生成了DataFrame
*/
DataFrame df = sqlContext.createDataFrame(personRDD, Person.class);
df.show();
df.registerTempTable(“person”);
sqlContext.sql(“select name from person where id = 2”).show();
/**
* 将DataFrame转成JavaRDD
* 注意：
* 1.可以使用row.getInt(0),row.getString(1)…通过下标获取返回Row类型的数据，但是要注意列顺序问题—不常用
* 2.可以使用row.getAs(“列名”)来获取对应的列值。
*
*/
JavaRDD<Row> javaRDD = df.javaRDD();
JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Person call(Row row) throws Exception {
Person p = new Person();
//p.setId(row.getString(1));
//p.setName(row.getString(2));
//p.setAge(row.getInt(0));
p.setId((String)row.getAs(“id”));
p.setName((String)row.getAs(“name”));
p.setAge((Integer)row.getAs(“age”));
return p;
}
});
map.foreach(new VoidFunction<Person>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public void call(Person t) throws Exception {
System.out.println(t);
}
});
sc.stop();
scala:
val conf = new SparkConf()
conf.setMaster(“local”).setAppName(“rddreflect”)
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val lineRDD = sc.textFile(“./sparksql/person.txt”)
/**
* 将RDD隐式转换成DataFrame
*/
import sqlContext.implicits._
val personRDD = lineRDD.map { x => {
val person = Person(x.split(“,”)(0),x.split(“,”)(1),Integer.valueOf(x.split(“,”)(2)))
person
} }
val df = personRDD.toDF();
df.show()
/**
* 将DataFrame转换成PersonRDD
*/
val rdd = df.rdd
val result = rdd.map { x => {
Person(x.getAs(“id”),x.getAs(“name”),x.getAs(“age”))
} }
result.foreach { println}
sc.stop()
动态创建Schema将非json格式的RDD转换成DataFrame：
java：
SparkConf conf = new SparkConf();
conf.setMaster(“local”).setAppName(“rddStruct”);
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile(“./sparksql/person.txt”);
/**
* 转换成Row类型的RDD
*/
JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Row call(String s) throws Exception {
return RowFactory.create(
String.valueOf(s.split(“,”)[0]),
String.valueOf(s.split(“,”)[1]),
Integer.valueOf(s.split(“,”)[2])
);
}
});
/**
* 动态构建DataFrame中的元数据，一般来说这里的字段可以来源自字符串，也可以来源于外部数据库
*/
List<StructField> asList =Arrays.asList(
DataTypes.createStructField(“id”, DataTypes.StringType, true),
DataTypes.createStructField(“name”, DataTypes.StringType, true),
DataTypes.createStructField(“age”, DataTypes.IntegerType, true)
);
StructType schema = DataTypes.createStructType(asList);
DataFrame df = sqlContext.createDataFrame(rowRDD, schema);
df.show();
sc.stop();
scala：
val conf = new SparkConf()
conf.setMaster(“local”).setAppName(“rddStruct”)
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val lineRDD = sc.textFile(“./sparksql/person.txt”)
val rowRDD = lineRDD.map { x => {
val split = x.split(“,”)
RowFactory.create(split(0),split(1),Integer.valueOf(split(2)))
} }
val schema = StructType(List(
StructField(“id”,StringType,true),
StructField(“name”,StringType,true),
StructField(“age”,IntegerType,true)
))
val df = sqlContext.createDataFrame(rowRDD, schema)
df.show()
df.printSchema()
sc.stop()
读取parquet文件创建DataFrame
注意：
可以将DataFrame存储成parquet文件。保存成parquet文件的方式有两种。
df.write().mode(SaveMode.Overwrite)format(“parquet”)
.save(“./sparksql/parquet”);
df.write().mode(SaveMode.Overwrite).parquet(“./sparksql/parquet”);
SaveMode指定文件保存时的模式。
Overwrite：覆盖
Append：追加
ErrorIfExists：如果存在就报错
Ignore：如果存在就忽略
java:
SparkConf conf = new SparkConf();
conf.setMaster(“local”).setAppName(“parquet”);
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> jsonRDD = sc.textFile(“sparksql/json”);
DataFrame df = sqlContext.read().json(jsonRDD);
/**
* 将DataFrame保存成parquet文件，SaveMode指定存储文件时的保存模式
* 保存成parquet文件有以下两种方式：
*/
df.write().mode(SaveMode.Overwrite).format(“parquet”).save(“./sparksql/parquet”);
df.write().mode(SaveMode.Overwrite).parquet(“./sparksql/parquet”);
df.show();
/**
* 加载parquet文件成DataFrame
* 加载parquet文件有以下两种方式：
*/
DataFrame load = sqlContext.read().format(“parquet”).load(“./sparksql/parquet”);
load = sqlContext.read().parquet(“./sparksql/parquet”);
load.show();
sc.stop();
scala：
val conf = new SparkConf()
conf.setMaster(“local”).setAppName(“parquet”)
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val jsonRDD = sc.textFile(“sparksql/json”)
val df = sqlContext.read.json(jsonRDD)
df.show()
/**
* 将DF保存为parquet文件
*/
df.write.mode(SaveMode.Overwrite).format(“parquet”).save(“./sparksql/parquet”)
df.write.mode(SaveMode.Overwrite).parquet(“./sparksql/parquet”)
/**
* 读取parquet文件
*/
var result = sqlContext.read.parquet(“./sparksql/parquet”)
result = sqlContext.read.format(“parquet”).load(“./sparksql/parquet”)
result.show()
sc.stop()
读取JDBC中的数据创建DataFrame(MySql为例)
两种方式创建DataFrame
java:
SparkConf conf = new SparkConf();
conf.setMaster(“local”).setAppName(“mysql”);
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
/**
* 第一种方式读取MySql数据库表，加载为DataFrame
*/
Map<String, String> options = new HashMap<String,String>();
options.put(“url”, “jdbc:mysql://192.168.179.4:3306/spark”);
options.put(“driver”, “com.mysql.jdbc.Driver”);
options.put(“user”, “root”);
options.put(“password”, “123456”);
options.put(“dbtable”, “person”);
DataFrame person = sqlContext.read().format(“jdbc”).options(options).load();
person.show();
person.registerTempTable(“person”);
/**
* 第二种方式读取MySql数据表加载为DataFrame
*/
DataFrameReader reader = sqlContext.read().format(“jdbc”);
reader.option(“url”, “jdbc:mysql://192.168.179.4:3306/spark”);
reader.option(“driver”, “com.mysql.jdbc.Driver”);
reader.option(“user”, “root”);
reader.option(“password”, “123456”);
reader.option(“dbtable”, “score”);
DataFrame score = reader.load();
score.show();
score.registerTempTable(“score”);
DataFrame result =
sqlContext.sql(“select person.id,person.name,score.score from person,score where person.name = score.name”);
result.show();
/**
* 将DataFrame结果保存到Mysql中
*/
Properties properties = new Properties();
properties.setProperty(“user”, “root”);
properties.setProperty(“password”, “123456”);
result.write().mode(SaveMode.Overwrite).jdbc(“jdbc:mysql://192.168.179.4:3306/spark”, “result”, properties);
sc.stop();
scala:
val conf = new SparkConf()
conf.setMaster(“local”).setAppName(“mysql”)
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
/**
* 第一种方式读取Mysql数据库表创建DF
*/
val options = new HashMap[String,String]();
options.put(“url”, “jdbc:mysql://192.168.179.4:3306/spark”)
options.put(“driver”,”com.mysql.jdbc.Driver”)
options.put(“user”,”root”)
options.put(“password”, “123456”)
options.put(“dbtable”,”person”)
val person = sqlContext.read.format(“jdbc”).options(options).load()
person.show()
person.registerTempTable(“person”)
/**
* 第二种方式读取Mysql数据库表创建DF
*/
val reader = sqlContext.read.format(“jdbc”)
reader.option(“url”, “jdbc:mysql://192.168.179.4:3306/spark”)
reader.option(“driver”,”com.mysql.jdbc.Driver”)
reader.option(“user”,”root”)
reader.option(“password”,”123456″)
reader.option(“dbtable”, “score”)
val score = reader.load()
score.show()
score.registerTempTable(“score”)
val result = sqlContext.sql(“select person.id,person.name,score.score from person,score where person.name = score.name”)
result.show()
/**
* 将数据写入到Mysql表中
*/
val properties = new Properties()
properties.setProperty(“user”, “root”)
properties.setProperty(“password”, “123456”)
result.write.mode(SaveMode.Append).jdbc(“jdbc:mysql://192.168.179.4:3306/spark”, “result”, properties)
sc.stop()
读取Hive中的数据加载成DataFrame：
HiveContext是SQLContext的子类，连接Hive建议使用HiveContext。
由于本地没有Hive环境，要提交到集群运行，提交命令：
./spark-submit
–master spark://node1:7077,node2:7077
–executor-cores 1
–executor-memory 2G
–total-executor-cores 1
–class com.bjsxt.sparksql.dataframe.CreateDFFromHive
/root/test/HiveTest.jar
java:
SparkConf conf = new SparkConf();
conf.setAppName(“hive”);
JavaSparkContext sc = new JavaSparkContext(conf);
//HiveContext是SQLContext的子类。
HiveContext hiveContext = new HiveContext(sc);
hiveContext.sql(“USE spark”);
hiveContext.sql(“DROP TABLE IF EXISTS student_infos”);
//在hive中创建student_infos表
hiveContext.sql(“CREATE TABLE IF NOT EXISTS student_infos (name STRING,age INT) row format delimited fields terminated by \t “);
hiveContext.sql(“load data local inpath /root/test/student_infos into table student_infos”);
hiveContext.sql(“DROP TABLE IF EXISTS student_scores”);
hiveContext.sql(“CREATE TABLE IF NOT EXISTS student_scores (name STRING, score INT) row format delimited fields terminated by \t”);
hiveContext.sql(“LOAD DATA ”
+ “LOCAL INPATH /root/test/student_scores”
+ “INTO TABLE student_scores”);
/**
* 查询表生成DataFrame
*/
DataFrame goodStudentsDF = hiveContext.sql(“SELECT si.name, si.age, ss.score ”
+ “FROM student_infos si ”
+ “JOIN student_scores ss ”
+ “ON si.name=ss.name ”
+ “WHERE ss.score>=80”);
hiveContext.sql(“DROP TABLE IF EXISTS good_student_infos”);
goodStudentsDF.registerTempTable(“goodstudent”);
DataFrame result = hiveContext.sql(“select * from goodstudent”);
result.show();
/**
* 将结果保存到hive表 good_student_infos
*/
goodStudentsDF.write().mode(SaveMode.Overwrite).saveAsTable(“good_student_infos”);
Row[] goodStudentRows = hiveContext.table(“good_student_infos”).collect();
for(Row goodStudentRow : goodStudentRows) {
System.out.println(goodStudentRow);
}
sc.stop();
scala:
val conf = new SparkConf()
conf.setAppName(“HiveSource”)
val sc = new SparkContext(conf)
/**
* HiveContext是SQLContext的子类。
*/
val hiveContext = new HiveContext(sc)
hiveContext.sql(“use spark”)
hiveContext.sql(“drop table if exists student_infos”)
hiveContext.sql(“create table if not exists student_infos (name string,age int) row format delimited fields terminated by \t”)
hiveContext.sql(“load data local inpath /root/test/student_infos into table student_infos”)
hiveContext.sql(“drop table if exists student_scores”)
hiveContext.sql(“create table if not exists student_scores (name string,score int) row format delimited fields terminated by \t”)
hiveContext.sql(“load data local inpath /root/test/student_scores into table student_scores”)
val df = hiveContext.sql(“select si.name,si.age,ss.score from student_infos si,student_scores ss where si.name = ss.name”)
hiveContext.sql(“drop table if exists good_student_infos”)
/**
* 将结果写入到hive表中
*/
df.write.mode(SaveMode.Overwrite).saveAsTable(“good_student_infos”)
sc.stop()
暂无评论内容