8.2 Spark外部数据源

快来打我* 2022-12-09 11:57 117阅读 0赞

### 文章目录 ###

*   *   *  Apache Parquet 开荒
 *  Data Sources
 *   *  1 通用的加载和保存方式
     *   *  1.1 加载数据
         *  1.2 保存数据
     *  2 读取文件实例
     *   *   *  2.1 Json
             *  2.2 Parquet
             *  2.3 CSV
             *  2.4 MySQL
             *  2.5 Hive

### Apache Parquet 开荒 ###

[Apache Parquet][]

Parquet 是 Hadoop 生态圈中主流的列式存储格式，最早是由 Twitter 和 Cloudera 合作开发，2015 年 5 月从 Apache 孵化器里毕业成为 Apache 顶级项目。

有这样一句话流传： HDFS 是大数据时代文件系统的事实标准，Parquet 就是大数据时代存储格式的事实标准。

**Parquet与语言 框架无关**

Parquet 是与语言无关的，而且不与任何一种数据处理框架绑定在一起，适配多种语言和组件，能够与 Parquet 适配的查询引擎包括 Hive, Impala, Pig, Presto, Drill, Tajo, HAWQ, IBM Big SQL等，计算框架包括 MapReduce, Spark, Cascading, Crunch, Scalding, Kite 等，数据模型包括 Avro, Thrift, Protocol Buffer, POJOs 等。

**Parquet文件格式**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]  
Parquet文件的每个文件块负责存储一个行组， 行组 由列块组成 ，一个列块负责存储一列数据，每个列块数据以页为单位

**Parquet的原子类型**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]  
**Parquet的逻辑类型**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]

# Data Sources #

[Spark Data Sources][]

SparkSQL 提供了通用的API去读不同格式的数据，默认保存为Parquet格式

## 1 通用的加载和保存方式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]

### 1.1 加载数据 ###

scala> spark.read.format("…")[.option("…")].load("…")

*  format("…"):指定加载数据的类型，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”。
 *  load("…"):在"csv"、“jdbc”、“json”、“orc”、"parquet"和"textFile"格式下需要传入加载数据的路径
 *  option("…"):在"jdbc"格式下需要传入JDBC相应参数，url、user、password和dbtable  
    我们前面都是使用read API 先把文件加载到 DataFrame然后再查询，其实，我们也可以直接在文件上进行查询: 文件格式.`文件路径`

scala>spark.sql("select * from json.`/opt/module/data/user.json`").show

### 1.2 保存数据 ###

## 2 读取文件实例 ##

#### 2.1 Json ####

JavaScript Object Notation

/*
        * 1 JSON
        * */
        import spark.implicits._
        //读取数据
        val JSON = spark.read.json("D:\\workplace\\ifeng-Spark\\ifeng-Spark-core\\data\\employees.json")
        //创建临时表
        val JsonView = JSON.createOrReplaceTempView("emp")
        //查询数据
        spark.sql("select name from emp where salary > 3500").show()

#### 2.2 Parquet ####

Spark SQL的默认数据源为Parquet 格式，Parquet是一种有效存储列式格式

数据源为Parquet文件时，不需要使用format

/*
        * 2 Parquet
        * SparkSQL 默认的数据源格式，Parquet是一种能够有效存储嵌套数据 的列存储格式
        * 数据源为Parquet时，不需要使用format
        * */
        //读取Parquet
        val ParquetRDD = spark.read.load("D:\\workplace\\ifeng-Spark\\ifeng-Spark-core\\data\\users.parquet")
        // careteOrRepalceTempView
        val ViewParquet = ParquetRDD.createOrReplaceTempView("color")
        // 
        spark.sql("select * from color").show()

#### 2.3 CSV ####

Spark SQL 可以配置CSV文件的列表信息，读取CSV文件第一行设置为数据列

/*
        *  3 CSV
        *
        * */
        val CSVRDD = spark.read.format("csv")
          .option("sep", ";") //指定分隔符
          .option("inferSchema",true) //推测数据的类型，时间字段等
          .option("header", true) //保留头信息
          .load("D:\\workplace\\ifeng-Spark\\ifeng-Spark-core\\data\\people.csv")
    
    //    CSVRDD.show()

[对于inferSchema的stackoverflow][inferSchema_stackoverflow]

#### 2.4 MySQL ####

**方式1：通用的load方法读取**

//方式1：通用的load方法读取
        val MySQLRDD1 = spark.read.format("jdbc")
          .option("url", "jdbc:mysql://10.103.66.88:3306/lenovosbom")
          .option("driver", "com.mysql.jdbc.Driver")
          .option("user", "xingwj1")
          .option("password", "xingwj1")
          //.option("dbtable", "ec")
          .option("query", "select * from ec")
          .load()

**方式2：load方式 配置集中放置在一起**

//
        val mySQLRDD2 = spark.read.format("jdbc")
          .options(Map("url" -> "jdbc:mysql://10.103.66.88:3306/lenovosbom?user=xingwj1&password=xingwj1",
                "dbtable" -> "user", "driver" -> "com.mysql.jdbc.Driver","dbtable" -> "ec")).load().show

**方式3：JDBC方式获取**

//方式3：JDBC方式获取
            val props: Properties = new Properties()
            props.setProperty("user", "xingwj1")
            props.setProperty("password", "xingwj1")
            val df: DataFrame = spark.read.jdbc("jdbc:mysql://10.103.66.88:3306/lenovosbom", "ec", props)
            df.show

--------------------

**保存数据**

// Saving data to a JDBC source
    MySQLRDD.write
      .format("jdbc")
      .option("url", "jdbc:postgresql:dbserver")
      .option("dbtable", "schema.tablename")
      .option("user", "username")
      .option("password", "password")
      .save()

#### 2.5 Hive ####

Spark SQL

1.  导入依赖

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_2.12</artifactId>
        <version>${
        spark.version}</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-exec</artifactId>
        <version>3.1.2</version>
    </dependency>

val spark: SparkSession = SparkSession.builder().master("local[*]").getOrCreate()
    
        spark.sql("show databases").show()

[Apache Parquet]: http://parquet.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/dbfdbee820e04060befb4f701d525db7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/d5da5de312734bd7b0ed254b81dc7af9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/ef1d6838ac524bb3a0f171f7f79a505c.png
[Spark Data Sources]: http://spark.apache.org/docs/latest/sql-data-sources.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/861ff814f00149d085daf20eb3170949.png
[inferSchema_stackoverflow]: https://stackoverflow.com/questions/31719575/inferschema-in-spark-csv-package