zoukankan      html  css  js  c++  java
  • 寒假学习(十)

    1.Spark SQL 基本操作
    { "id":1 ,"name":" Ella","age":36 }
    { "id":2,"name":"Bob","age":29 }
    { "id":3 ,"name":"Jack","age":29 }
    { "id":4 ,"name":"Jim","age":28 }
    { "id":5 ,"name":"Damon" }
    { "id":5 ,"name":"Damon" }
    首先为employee.json 创建 DataFrame,并写出 Scala 语句完成下列操作:
    创建 DataFrame
     
    scala> import org.apache.spark.sql.SparkSession
    scala> val spark=SparkSession.builder().getOrCreate()
    scala> import spark.implicits._
    scala> val df = spark.read.json("file:///usr/local/spark/employee.json")
     
    (1) 查询 DataFrame 的所有数据
    scala> df.show()
    (2) 查询所有数据,并去除重复的数据
    scala> df.distinct().show()
    (3) 查询所有数据,打印时去除 id 字段
    scala> df.drop("id").show()
    (4) 筛选 age>20 的记录
    scala> df.filter(df("age") > 30 ).show()
    (5) 将数据按 name 分组
    scala> df.groupBy("name").count().show()
    (6) 将数据按 name 升序排列
    scala> df.sort(df("name").asc).show()
    (7) 取出前 3 行数据
    scala> df.take(3) 或 scala> df.head(3)
    (8) 查询所有记录的 name 列,并为其取别名为 username
    scala> df.select(df("name").as("username")).show()
    (9) 查询年龄 age 的平均值
    scala> df.agg("age"->"avg")
    (10) 查询年龄 age 的最小值
    scala> df.agg("age"->"min")
    2.编程实现将 RDD 转换为 DataFrame
    源文件内容如下(包含 id,name,age),将数据复制保存到 ubuntu 系统/usr/local/spark 下,
    命名为 employee.txt,实现从 RDD 转换得到 DataFrame,并按 id:1,name:Ella,age:36 的格式
    打印出 DataFrame 的所有数据。请写出程序代码。(任选一种方法即可)
    1,Ella,36
    2,Bob,29
    3,Jack,29

    假设当前目录为/usr/local/spark/mycode/rddtodf,在当前目录下新建一个目录 mkdir -p

    src/main/scala ,然后在目录 /usr/local/spark/mycode/rddtodf/src/main/scala 下 新 建 一 个
    rddtodf.scala,利用反射来推断包含特定类型对象的 RDD 的 schema,适用对已知数据结构的 RDD
    转换;
    import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
    import org.apache.spark.sql.Encoder
    import spark.implicits._
    object RDDtoDF {
     def main(args: Array[String]) {
    case class Employee(id:Long,name: String, age: Long)
    val employeeDF = 
    spark.sparkContext.textFile("file:///usr/local/spark/employee.txt").map(_.split(",")).map(at
    tributes => Employee(attributes(0).trim.toInt,attributes(1), attributes(2).trim.toInt)).toDF()
    employeeDF.createOrReplaceTempView("employee")
    val employeeRDD = spark.sql("select id,name,age from employee")
    employeeRDD.map(t => "id:"+t(0)+","+"name:"+t(1)+","+"age:"+t(2)).show()
     } }
  • 相关阅读:
    548
    zookeeper的异常处理(Disconnected, SyncConnected, Expired)
    ZOJ 2588 Burning Bridges(求含重边的无向连通图的割边)
    SQL Server 中VARCHAR(MAX)变量赋值引起的性能问题。
    查看mysql数据库版本方法总结
    Linux mysql 5.6: ERROR 1045 (28000): Access denied for user 'root'@'localhost' (using password: NO)
    Ignite用户配置管理介绍
    ORACLE动态采样分析
    SQL SERVER 2012链接到SQL SERVER 2000的问题解决案例
    SQL SERVER 2005删除维护作业报错:The DELETE statement conflicted with the REFERENCE constraint "FK_subplan_job_id"
  • 原文地址:https://www.cnblogs.com/sonofdemon/p/12289410.html
Copyright © 2011-2022 走看看