码迷，mamicode.com
首页 > 其他好文 > 详细

Spark_总结四

时间：2017-03-05 17:30:42      阅读：313      评论：0      收藏：0      [点我收藏+]

标签：gty   配置文件   group by   结合   情况   对象   conf   hadoop1   default   

 

Spark_总结四

1.Spark SQL  

技术分享

    Spark SQL 和 Hive on Spark 两者的区别？
        spark on hive：hive只是作为元数据存储的角色，解析，优化，执行都是spark做的    
        hive on spark： hive既作为存储的角色，又作为计算角色的一部分，hive将sql解析Spark任务，底层是Spark引擎（hive2.0以后推荐使用Spark引擎，转化为Spark任务，hvie2.0以前都是转化为MR任务）
            
    Spark SQL 转化的过程（底层架构）
技术分享
 
【SQL/HQL-->解析器-->分析器-->优化器-->CostModel消耗模型（选出消耗最低的，就是效率最高的），最终将传入的SQL转换为RDD的计算】
 
须知：
        若想使用SparkSQL必须创建SQLContext 必须是传入SparkContext 不能是SparkConf
技术分享
 
1.DataFrame与RDD的区别？   ||   什么是DataFrame?

技术分享

区别：
      Spark core是基于RDD的编程，Spark SQL是基于DataFrame的编程，DataFrame的底层就是封装的RDD，只不过DataFrame底层RDD的泛型是ROW（DataFrame <==> RDD<ROW>），另外，DataFrame中有对列的描述，但是RDD没有对列的描述。      
What is DataFrame：
      DataFrame 与 RDD 类似，DataFrame 是一个分布式数据容器，更像传统数据库的二维表格，除了数据以外，还掌握数据的结构信息（比如对列的描述）， 即 schema。同时，与 Hive 类似，DataFrame 也支持嵌套数据类型（struct、 array 和 map）。 从 API 易用性的角度上 看，DataFrameAPI 提供的是一套高层的关系操作，比函数式的 RDDAPI 要更加友好，门槛更低。 

3.创建DataFrame的来源和方式   ||   如何对DataFrame中封装的数据进行操作？

3.1创建DataFrame的来源和方式
技术分享
 

3.2如何对DataFrame中封装的数据进行操作？

   当我们的DataFrame构建好之后，里面封装了我们的数据，需要对数据进行操作即对DataFrame进行操作，有两种方式
3.2.1   通过方法
        sqlContext.read()    返回DataFrameReader对象
        sqlContext.read().json("student.json")   读取一个json文件（这个json文件中的内容不能是嵌套的）读进来变成DataFrame,
        df.select("age").show()，如果没有show,这个程序就不会执行，这个show就类似与Spark中Action类型的算子，触发执行
示例代码：
  1. package com.hzf.spark.exercise;
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaSparkContext;
  5. import org.apache.spark.sql.DataFrame;
  6. import org.apache.spark.sql.SQLContext;
  8. publicclassTestSparkSQL02{
  9. publicstaticvoid main(String[] args){
  10. SparkConf conf =newSparkConf().setAppName("DataFrameOps").setMaster("local");
  12. JavaSparkContext sc =newJavaSparkContext(conf);
  13. SQLContext sqlContext =newSQLContext(sc);
  15. DataFrame df = sqlContext.read().json("people.json");
  18. /*
  19. * 操作DataFrame的第一种方式
  20. * */
  21. //类似 SQL的select from table;
  22. df.show();
  23. //desc table
  24. df.printSchema();
  26. //select age from table;
  27. df.select("age").show();
  28. //select name from table;
  29. df.select("name").show();
  30. //select name,age+10 from table;
  31. df.select(df.col("name"),df.col("age").plus(10)).show();
  32. //select * from table where age > 20
  33. df.filter(df.col("age").gt(20)).show();
  34. }
  35. }
result:
技术分享
 
3.2.2   通过注册临时表，传入SQL语句（推荐使用）
示例代码：
  1. package com.hzf.spark.exercise;
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaSparkContext;
  5. import org.apache.spark.sql.DataFrame;
  6. import org.apache.spark.sql.SQLContext;
  8. publicclassTestSparkSQL01{
  9. publicstaticvoid main(String[] args){
  10. SparkConf conf =newSparkConf().setAppName("DataFrameOps").setMaster("local");
  12. JavaSparkContext sc =newJavaSparkContext(conf);
  13. SQLContext sqlContext =newSQLContext(sc);
  15. DataFrame df = sqlContext.read().json("people.json");
  17. //将DataFrame中封装的数据注册为一张临时表，对临时表进行sql操作
  18. df.registerTempTable("people");
  19. DataFrame sql = sqlContext.sql("SELECT * FROM people WHERE age IS NOT NULL");
  20. sql.show();
  21. }
  22. }
result:
技术分享
 
3.3创建DataFrame的几种方式，来源（json，jsonRDD，parquet，非json格式，mysql）
 
<1>读取Json格式文件-->DataFrame：Json 文件中不能有嵌套的格式
      加载json格式文件-->DataFrame有两种方式：
           方式一：DataFrame df = sqlContext.read().format("json").load("people.json");
           方式二：DataFrame df = sqlContext.read().json("people.json");
数据集：
技术分享
示例代码：
  1. package com.bjsxt.java.spark.sql.json;
  3. import java.util.ArrayList;
  4. import java.util.List;
  6. import org.apache.spark.SparkConf;
  7. import org.apache.spark.SparkContext;
  8. import org.apache.spark.api.java.JavaPairRDD;
  9. import org.apache.spark.api.java.JavaRDD;
  10. import org.apache.spark.api.java.JavaSparkContext;
  11. import org.apache.spark.api.java.function.Function;
  12. import org.apache.spark.api.java.function.PairFunction;
  13. import org.apache.spark.sql.DataFrame;
  14. import org.apache.spark.sql.Row;
  15. import org.apache.spark.sql.RowFactory;
  16. import org.apache.spark.sql.SQLContext;
  17. import org.apache.spark.sql.types.DataTypes;
  18. import org.apache.spark.sql.types.StructField;
  19. import org.apache.spark.sql.types.StructType;
  21. import scala.Tuple2;
  23. /**
  24. * JSON数据源
  25. * @author Administrator
  26. *
  27. */
  28. publicclassJSONDataSource{
  30. publicstaticvoid main(String[] args){
  31. SparkConf conf =newSparkConf()
  32. .setAppName("JSONDataSource")
  33. // .set("spark.default.parallelism", "100")
  34. .setMaster("local");
  35. JavaSparkContext sc =newJavaSparkContext(conf);
  36. SQLContext sqlContext =newSQLContext(sc);
  38. DataFrame studentScoresDF = sqlContext.read().json("student.json");
  40. studentScoresDF.registerTempTable("student_scores");
  41. DataFrame goodStudentScoresDF = sqlContext.sql(
  42. "select name,count(score) from student_scores where score>=80 group by name");
  44. List<String> goodStudentNames = goodStudentScoresDF.javaRDD().map(newFunction<Row,String>(){
  45. privatestaticfinallong serialVersionUID =1L;
  47. @Override
  48. publicString call(Row row)throwsException{
  49. return row.getString(0);
  50. }
  52. }).collect();
  54. for(String str: goodStudentNames){
  55. System.out.println(str);
  56. }
  57. }
  58. }
result:
技术分享
 
<2>jsonRDD-->DataFrame
技术分享
 
<3>读取Parquet格式文件-->DataFrame：自动推测分区，合并 Schema。
经验：将Spark中的文本转换为Parquet以提升性能

    parquet是一个基于列的存储格式，列式存储布局可以加速查询，因为它只检查所有需要的列并对它们的值执行计算，因此只读取一个数据文件或表的小部分数据。Parquet 还支持灵活的压缩选项，因此可以显著减少磁盘上的存储。

   如果在 HDFS 上拥有基于文本的数据文件或表，而且正在使用 Spark SQL 对它们执行查询，那么强烈推荐将文本数据文件转换为 Parquet 数据文件，以实现性能和存储收益。当然，转换需要时间，但查询性能的提升在某些情况下可能达到 30 倍或更高，存储的节省可高达 75%！

parquet的压缩比高，将一个普通的文本转化为parquet格式，如何去转？
       val lineRDD = sc.textFile()
       DF.save(parquet) //将RDD转化为DF
parquet操作示例
   是否指定format--若存储时，指定format为json格式，那么则生成json格式文件，否则不指定format,默认文件以parquet形式进行存储 
测试一：指定format为json格式，存储在本地
测试数据:   top.txt
技术分享
测试代码

技术分享

测试结果
技术分享
 
技术分享
 
测试二：不指定format,那么文件默认以parquet形式进行存储，存储在本地
 
测试数据:   people.json
技术分享
测试代码
技术分享
测试结果
技术分享
技术分享
 
测试三：读取本地parquet存储格式的文件
测试代码
技术分享
测试结果
技术分享
 
测试四：读取hdfs上parquet形式的文件
技术分享
测试代码
技术分享
测试结果
技术分享
 
<4> RDD（非json格式变成DataFrame）
读取txt 文件-->DataFrame：从 txt 文件读取，然后转为 RDD，最后转为 DataFrame
                                       RDD 转为 DataFrame 有两种方式
                        (1)反射机制，
                              注意点：自定义的类一定要是 public，并且要实现序列化接口 Serializable，
                                             取数据的时候，在 JavaAPI 中会有顺序问题（因为 DataFrame 转为 RDD<Row> 的时候，会进行一次字典排序改变 Row 的位置，而Scala 的 API 则没有这个问题）
                        (2)动态创建 Schema，先将 RDD 中的每一行类型变 为 RDD<Row> 类型，然后创建 DataFrame 的元数据-->构建 StructType，用于最后 DataFrame 元数据的描述，基于现有的 StructType 以及 RDD<Row> 来构造 DataFrame。(如果列的信息比较长可以存到数据库里) 
<4.1>反射机制
数据
技术分享
示例代码：
自定义类
技术分享
  1. package com.bjsxt.java.spark.sql.createdf;
  3. import java.util.List;
  5. import org.apache.spark.SparkConf;
  6. import org.apache.spark.api.java.JavaRDD;
  7. import org.apache.spark.api.java.JavaSparkContext;
  8. import org.apache.spark.api.java.function.Function;
  9. import org.apache.spark.sql.DataFrame;
  10. import org.apache.spark.sql.Row;
  11. import org.apache.spark.sql.SQLContext;
  13. /**
  14. * 使用反射的方式将RDD转换成为DataFrame
  15. * 1、自定义的类必须是public
  16. * 2、自定义的类必须是可序列化的
  17. * 3、RDD转成DataFrame的时候，他会根据自定义类中的字段名进行排序。
  18. * @author zfg
  19. *
  20. */
  22. publicclass RDD2DataFrameByReflection {
  24. publicstaticvoid main(String[] args){
  25. SparkConf conf =newSparkConf().setMaster("local").setAppName("RDD2DataFrameByReflection");
  26. JavaSparkContext sc =newJavaSparkContext(conf);
  27. SQLContext sqlcontext =newSQLContext(sc);
  29. JavaRDD<String> lines = sc.textFile("Peoples.txt");
  30. JavaRDD<Person> personsRdd = lines.map(newFunction<String,Person>(){
  32. privatestaticfinallong serialVersionUID =1L;
  34. @Override
  35. publicPerson call(String line)throwsException{
  36. String[] split = line.split(",");
  37. Person p =newPerson();
  38. p.setId(Integer.valueOf(split[0].trim()));
  39. p.setName(split[1]);
  40. p.setAge(Integer.valueOf(split[2].trim()));
  41. return p;
  42. }
  43. });
  45. //传入进去Person.class的时候，sqlContext是通过反射的方式创建DataFrame
  46. //在底层通过反射的方式或得Person的所有field，结合RDD本身，就生成了DataFrame
  47. DataFrame df = sqlcontext.createDataFrame(personsRdd,Person.class);
  49. //命名table的名字为person
  50. df.registerTempTable("personTable");
  52. DataFrame resultDataFrame = sqlcontext.sql("select * from personTable where age > 7");
  53. resultDataFrame.show();
  55. //将df转成rdd
  56. JavaRDD<Row> resultRDD = resultDataFrame.javaRDD();
  57. JavaRDD<Person> result = resultRDD.map(newFunction<Row,Person>(){
  59. privatestaticfinallong serialVersionUID =1L;
  61. @Override
  62. publicPerson call(Row row)throwsException{
  63. Person p =newPerson();
  64. p.setAge(row.getInt(0));
  65. p.setId(row.getInt(1));
  66. p.setName(row.getString(2));
  67. return p;
  68. }
  69. });
  71. List<Person> personList = result.collect();
  73. for(Person person : personList){
  74. System.out.println(person.toString());
  75. }
  76. }
  77. }
result：
技术分享
技术分享

<4.2>动态创建Schema方式
数据
技术分享
示例代码：
  1. package com.bjsxt.java.spark.sql.createdf;
  3. import java.util.ArrayList;
  4. import java.util.List;
  6. import org.apache.spark.SparkConf;
  7. import org.apache.spark.api.java.JavaRDD;
  8. import org.apache.spark.api.java.JavaSparkContext;
  9. import org.apache.spark.api.java.function.Function;
  10. import org.apache.spark.sql.DataFrame;
  11. import org.apache.spark.sql.Row;
  12. import org.apache.spark.sql.RowFactory;
  13. import org.apache.spark.sql.SQLContext;
  14. import org.apache.spark.sql.types.DataTypes;
  15. import org.apache.spark.sql.types.StructField;
  16. import org.apache.spark.sql.types.StructType;
  19. publicclass RDD2DataFrameByProgrammatically {
  20. publicstaticvoid main(String[] args){
  21. SparkConf conf =newSparkConf().setMaster("local").setAppName("RDD2DataFrameByReflection");
  22. JavaSparkContext sc =newJavaSparkContext(conf);
  23. SQLContext sqlcontext =newSQLContext(sc);
  24. /**
  25. * 在RDD的基础上创建类型为Row的RDD
  26. */
  27. JavaRDD<String> lines = sc.textFile("Peoples.txt");
  28. JavaRDD<Row> rowRDD = lines.map(newFunction<String,Row>(){
  30. privatestaticfinallong serialVersionUID =1L;
  32. @Override
  33. publicRow call(String line)throwsException{
  34. String[] split = line.split(",");
  35. returnRowFactory.create(Integer.valueOf(split[0]),split[1],Integer.valueOf(split[2]));
  36. }
  37. });
  39. /**
  40. * 动态构造DataFrame的元数据，一般而言，有多少列以及每列的具体类型可能来自于Json，也可能来自于DB
  41. */
  42. ArrayList<StructField> structFields =newArrayList<StructField>();
  43. structFields.add(DataTypes.createStructField("id",DataTypes.IntegerType,true));
  44. structFields.add(DataTypes.createStructField("name",DataTypes.StringType,true));
  45. structFields.add(DataTypes.createStructField("age",DataTypes.IntegerType,true));
  46. //构建StructType，用于最后DataFrame元数据的描述
  47. StructType schema =DataTypes.createStructType(structFields);
  49. /**
  50. * 基于已有的MetaData以及RDD<Row> 来构造DataFrame
  51. */
  52. DataFrame df = sqlcontext.createDataFrame(rowRDD, schema);
  54. /**
  55. *注册成为临时表以供后续的SQL操作查询
  56. */
  57. df.registerTempTable("persons");
  59. /**
  60. * 进行数据的多维度分析
  61. */
  62. DataFrame result = sqlcontext.sql("select * from persons where age > 7");
  63. result.show();
  65. /**
  66. * 对结果进行处理，包括由DataFrame转换成为RDD<Row>
  67. */
  68. List<Row> listRow = result.javaRDD().collect();
  69. for(Row row : listRow){
  70. System.out.println(row);
  71. }
  72. }
  73. }
result：
技术分享
技术分享

<5> 读取MySql 中表里的数据-->DataFrame
       Spark Build-in内置支持的json jdbc mysql,hive...如果数据库支持jdbc连接，Spark 就可以基于这个数据库尽行数据的处理
示例代码：
  1. package com.bjsxt.java.spark.sql.jdbc;
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaSparkContext;
  5. import org.apache.spark.sql.DataFrame;
  6. import org.apache.spark.sql.DataFrameReader;
  7. import org.apache.spark.sql.SQLContext;
  9. /**
  10. * JDBC数据源
  11. *
  12. * @author Administrator
  13. *
  14. */
  15. publicclassJDBCDataSource{
  17. publicstaticvoid main(String[] args){
  18. SparkConf conf =newSparkConf().setAppName("JDBCDataSource").setMaster("local");
  19. JavaSparkContext sc =newJavaSparkContext(conf);
  20. SQLContext sqlContext =newSQLContext(sc);
  22. // 方法1、分别将mysql中两张表的数据加载为DataFrame
  23. /*
  24. * Map<String, String> options = new HashMap<String, String>();
  25. * options.put("url", "jdbc:mysql://hadoop1:3306/testdb");
  26. * options.put("driver", "com.mysql.jdbc.Driver");
  27. * options.put("user","spark");
  28. * options.put("password", "spark2016");
  30. * options.put("dbtable", "student_info");
  31. * DataFrame studentInfosDF = sqlContext.read().format("jdbc").options(options).load();
  32. * options.put("dbtable", "student_score");
  33. * DataFrame studentScoresDF = sqlContext.read().format("jdbc") .options(options).load();
  34. */
  35. // 方法2、分别将mysql中两张表的数据加载为DataFrame
  36. DataFrameReader reader = sqlContext.read().format("jdbc");
  37. reader.option("url","jdbc:mysql://node4:3306/testdb");
  38. reader.option("driver","com.mysql.jdbc.Driver");
  39. reader.option("user","root");
  40. reader.option("password","123");
  42. reader.option("dbtable","student_info");
  43. DataFrame studentInfosDF = reader.load();
  44. reader.option("dbtable","student_score");
  45. DataFrame studentScoresDF = reader.load();
  47. // 将两个DataFrame转换为JavaPairRDD，执行join操作
  48. studentInfosDF.registerTempTable("studentInfos");
  49. studentScoresDF.registerTempTable("studentScores");
  51. String sql ="SELECT studentInfos.name,age,score "
  52. +" FROM studentInfos JOIN studentScores"
  53. +" ON (studentScores.name = studentInfos.name)"
  54. +" WHERE studentScores.score > 80";
  56. DataFrame sql2 = sqlContext.sql(sql);
  57. sql2.show();
  58. }
  59. }
result:
技术分享

4. 如何将DataFrame中的值写入到外部存储中去？

存储模式（SaveMode.Overwrite || Ignore || Append   ||   ErrorifExit）
<1> 读取本地json格式文件，并以json形式写入到hdfs（不指定format,默认是parquet）
测试代码
技术分享
测试结果
技术分享
技术分享
 
补充：
1.什么是下推过滤器？
       在join之前过滤，而不是join之后进行过滤
 
2.select * from table 在SparkSQL和Hive on MR中的区别?
      SparkSQL 中 select * from table 在spark中是要具体执行spark任务的，而在 Hive on MR 中 select * from table直接读取数据，所以SparkSQL 中执行select * from不一定比Hive on MR中的快
 
3.如何将一个DataFrame变成一个RDD？
    JavaRDD<ROW> rdd = resultFrame.javaRDD()
 

5.整合Spark和Hive?

    6.1Spark 目录下面的 conf 下放一个配置文件 hive-site.xml 文件。
    6.2在 hive 的服务端启动 MetaStore Server【因为 HiveContext 会用到 metastore 服务。(在 Spark-shell 里面使用 HiveContext 的时候，要记住导入 HiveContext）】（hive --service metastore）
    6.3启动hdfs【因为hive的数据是存在hdfs上的】和Spark集群（start-all.sh spark-start-all.sh）
    6.4进入Spark shell,测试Spark 和 Hive是否整合成功
  1. scala>import org.apache.spark.sql.hive.HiveContext
  2. scala>val hiveContext =newHiveContext(sc)
  3. scala>hiveContext.sql("show tables").show
    6.5整合测试（详见Spark_some配置），注意！将代码提交到Spark集群上运行时，需要将hdfs-site.xml拷贝到SPARK_HOME/conf下

6.SqlContext和HiveContext的关系？

       SQLcontext 是 HiveContext 的父类
       在集群中运行的时候用 HiveContext，可以基于 Hive 来操作 Hive 表，对源数据进行CRUD的操作。 
  
    
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Spark_总结四

标签：gty   配置文件   group by   结合   情况   对象   conf   hadoop1   default   

原文地址：http://www.cnblogs.com/haozhengfei/p/22bba3b1ef90cbfaf073eb44349c0757.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！