zoukankan      html  css  js  c++  java
  • Spark(十二)【SparkSql中数据读取和保存】

    一. 读取和保存说明

    SparkSQL提供了通用的保存数据和数据加载的方式,还提供了专用的方式

    读取:通用和专用

    保存

    保存有四种模式:
      默认: error :  输出目录存在就报错
            append:  向输出目录追加
            overwrite : 覆盖写
            ignore:  忽略,不写
    

    二. 数据格式

    1. Parquet

    Spark SQL的默认数据源为Parquet格式。Parquet是一种能够有效存储嵌套数据的列式存储格式。

    数据源为Parquet文件时,Spark SQL可以方便的执行所有的操作,不需要使用format。修改配置项spark.sql.sources.default,可修改默认数据源格式。

    读取

    val df = spark.read.load("examples/src/main/resources/users.parquet")
    

    保存

    //读取json文件格式
    var df = spark.read.json("/opt/module/data/input/people.json")
    //保存为parquet格式
    df.write.mode("append").save("/opt/module/data/output")
    

    2. Json

    Spark SQL 能够自动推测JSON数据集的结构,并将它加载为一个Dataset[Row]. 可以通过SparkSession.read.json()去加载JSON 文件。

    注意:Spark读取的JSON文件不是传统的JSON文件,每一行都应该是一个JSON串。

    数据格式:employees.json

    {"name":"Michael"}
    {"name":"Andy", "age":30}
    

    1)导入隐式转换

    import spark.implicits._
    

    2)读取Json文件

    //专用的读取
    val df1: DataFrame = sparkSession.read.json("input/employees.json")
    //通用读取
    val df: DataFrame = sparkSession.read.format("json").load("input/employees.json")
    

    3)保存为Json文件

        //导隐式包,转为DataSet
        import sparkSession.implicits.
        val ds: Dataset[Emp] = rdd.toDS()
        ds.write.mode("overwrite")json("output/emp.json")
    

    3. CSV

    CSV: 逗号作为字段分割符的文件
    tsv: ,tab作为字段分割符的文件

    读取

        // 通用的读取
        val df: DataFrame = sparkSession.read.format("csv").load("input/person.csv")
        // 专用的读
        val df1: DataFrame = sparkSession.read.csv("input/person.csv")
    

    保存

    CSV的参数可以到DataFrameReader 609行查看

    //DataFrame
    df1.write.option("sep",",").mode("overwrite").csv("output/csv")
    

    4. Mysql

    读取

        val props = new Properties()
        /*
            JDBC中能写什么参数,参考 JDBCOptions 223行
         */
        props.put("user","root")
        props.put("password","root")
        //库名
        val df: DataFrame = sparkSession.read.jdbc("jdbc:mysql://localhost:3306/spark_test", "tbl_user", props)
        // 全表查询    只显示前N条
        df.show()
        //指定查询
        df.createTempView("user")
        sparkSession.sql("select * from user where id > 5").show()
    

    保存

        val list = List(Emp("jack", 2222.22), Emp("jack1", 3222.22), Emp("jack2", 4222.22))
        val rdd: RDD[Emp] = sparkSession.sparkContext.makeRDD(list, 1)
        //导入隐式包
        import sparkSession.implicits._
        val ds: Dataset[Emp] = rdd.toDS()
        val props = new Properties()
        props.put("user","root")
        props.put("password","root")
        //  表名可以是已经存在的表t1,也可以是一张新表t1(用的多)  
        
        //专用的写
        ds.write.jdbc("jdbc:mysql://localhost:3306/0508","t1",props)
    
    
    
        // 通用的写
        ds.write.
          option("url","jdbc:mysql://localhost:3306/库名")
    		//表名
          .option("dbtable","t2")
          .option("user","root")
          .option("password","root")
          .mode("append")
          .format("jdbc").save()
    
  • 相关阅读:
    吴裕雄--天生自然ANDROID开发学习:1.9 Android程序签名打包
    吴裕雄--天生自然ANDROID开发学习:1.8 工程相关解析(各种文件,资源访问)
    html 上传预览图片
    git笔记
    iscroll 下拉刷新功能
    移动端页面 弹出框滚动,底部body锁定,不滚动 / 微信网页禁止回弹效果
    getElementsByClassName 兼容性
    登录页面-输入框清空按钮
    FireFox中iframe的返回上页问题
    ajax是异步的,异步取数据,如何能保证数据是存在的。
  • 原文地址:https://www.cnblogs.com/wh984763176/p/13497285.html
Copyright © 2011-2022 走看看