微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦!

7.Spark SQL

------------恢复内容开始------------

1.请分析Sparksql出现的原因,并简述Sparksql的起源与发展。

  Shark提供了类似于Hive的功能,与Hive不同的是,Shark把sql语句转换成Spark作业,而不是MAPreduce作业。为了实现Hive的兼容,Shark重用了Hive中的Hive sql解析、逻辑执行计划翻译、执行计划优化等逻辑。可以近似的认为,Sark仅仅将物理执行计划从Map Reduce作业替换成了Spark作业,也就是通过Hive的Hivesql解析功能,把Hive sql翻译成Spark上的RDD操作。Shark的出现,使得sql-on-Hadoop的性能比Hive有了10~100倍的提高。

2.简述RDD 和DataFrame的联系与区别?

  Rdd是分布式的Java对象的集合,但是,对象内部的结构对于RDD来说是不可知的。Data Frame是一种以RDD为基础的分布式数据集,提供详细的结构信息,就相当于关系数据库的一张表。

3.DataFrame的创建

  spark.read.text(url)

  spark.read.json(url) 

  spark.read.format("text").load("people.txt")

  spark.read.format("json").load("people.json")

  描述从不同文件类型生成DataFrame的区别。

  用相同的txt或json文件,同时创建RDD,比较RDD与DataFrame的区别

  区别:RDD是直接输出对象,DataFrame是以对象里面的的详细结构进行输出

4.PySpark-DataFrame各种常用操作

  4.1基于df的操作

    打印数据 df.show()认打印前20条数据

    打印概要 df.printSchema()

    查询总行数 df.count()

    df.head(3) #list类型,list中每个元素是Row类

    输出全部行 df.collect() #list类型,list中每个元素是Row类

    查询概况 df.describe().show()

    取列 的三种方式 df[‘name’], df.name, df[1]

    选择 df.select() 每个人的年龄+1

    筛选 df.filter() 20岁以上的人员信息

    筛选年龄为空的人员信息

    分组df.groupBy() 统计每个年龄的人数

    排序df.sortBy() 按年龄进行排序

 

 

  4.2基于spark.sql的操作:

    创建临时表虚拟表 df.registerTempTable('people')

    spark.sql执行sql语句 spark.sql('select name from people').show()

5.Pyspark中DataFrame与pandas中DataFrame

  分别从文件创建DataFrame 

 

 

 

 

  pandas中DataFrame转换为Pyspark中DataFrame

  Pyspark中DataFrame转换为pandas中DataFrame 

  从创建与操作上,比较两者的异同

    pandas DataFrame有索引生成,而spark DateFrame没有索引

 

 

6.从RDD转换得到DataFrame

  6.1 利用反射机制推断RDD模式

    创建RDD sc.textFile(url).map(),读文件,分割数据项

    每个RDD元素转换成 Row

    由Row-RDD转换到DataFrame

 

 

 

  6.2 使用编程方式定义RDD模式

    #下面生成“表头” 

      schemaString = "name age"
      fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split(" ")]
      schema = StructType(fields)

    #下面生成“表中的记录” 

      lines = spark.sparkContext.textFile("file:///home/hadoop/people.txt")
      parts = lines.map(lambda x :x.split(","))
      people = parts.map(lambda p:Row(p[0],p[1].strip()))

    #下面把“表头”和“表中的记录”拼装在一起

      schemaPeople = spark.createDataFrame(people, schema)

 

 

 

 

7. DataFrame的保存

  df.write.text(dir)

 

  df.write.json(dri)

  df.write.format("text").save(dir)

  df.write.format("json").save(dir)

  df.write.format("json").save(dir)

 

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 [email protected] 举报,一经查实,本站将立刻删除。

相关推荐