首页 > 代码库 > Spark_总结四

Spark_总结四

2024-09-06 23:46:31   215人阅读

 

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_总结四


联系
我们