Spark2.X CSV文件创建DataFrame

叁歲伎倆 2023-07-06 06:20 3阅读 0赞

### 一、需求分析 ###

将CSV文件转为DataFrame，其中CSV包括无头文件和有文件。

### 二、数据展示 ###

无头文件的CSV

1,张飞,21,北京,80
    2,关羽,23,北京,82
    3,赵云,20,上海,88
    4,刘备,26,上海,83
    5,曹操,30,深圳,90.8

有头文件的CSV

id,name,age,city,score
    1,张飞,21,北京,80
    2,关羽,23,北京,82
    3,赵云,20,上海,88
    4,刘备,26,上海,83
    5,曹操,30,深圳,90.8

### 三、无头文件的CSV分析 ###

package blog
    
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.sql.{DataFrame, SparkSession}
    import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
    
    /**
     * @author: 余辉  
     * @blog: https://blog.csdn.net/silentwolfyh
     * @create: 2019-12-29 16:28
     * @description:
     **/
    object DF07_CSV_NotHeader {
    
      Logger.getLogger("org").setLevel(Level.WARN)
      
      def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession.builder()
          .appName("SparkUtils")
          .master("local[*]")
          .getOrCreate()
    
        /***
         *
         * 没有头文件的csv文件，要生成标准DataFrame
         * 第一种方式：
         * 1、要设置参数 inferSchema 为true ，让它自动推断类型
         * 2、将我的DF，重新设置列名称 .toDF("id","age","name","city","score")
         *
         * 没有 设置参数 inferSchema ，默认给的类型为string
         * root
         * |-- _c0: string (nullable = true)
         * |-- _c1: string (nullable = true)
         * |-- _c2: string (nullable = true)
         * |-- _c3: string (nullable = true)
         * |-- _c4: string (nullable = true)
         *
         * 设置参数 inferSchema ，默认给的类型为程序自动推断出来的
         * root
         * |-- _c0: integer (nullable = true)
         * |-- _c1: string (nullable = true)
         * |-- _c2: integer (nullable = true)
         * |-- _c3: string (nullable = true)
         * |-- _c4: double (nullable = true)
         *
         * 第二种方式：
         * 1、设置它的Schema ， 字段类型都在 DataTypes 中
         */
    
        /** *
         * spark.read.schema(schema) 方法：schema的源码查看
         *
         * case class StructType(fields: Array[StructField])
         *
         * case class StructField(
         * name: String,
         * dataType: DataType,
         * nullable: Boolean = true,
         * metadata: Metadata = Metadata.empty)
         */
        val schema = StructType(Array(
          StructField("id",DataTypes.IntegerType),
          StructField("name",DataTypes.StringType),
          StructField("age",DataTypes.IntegerType),
          StructField("city",DataTypes.StringType),
          StructField("score",DataTypes.DoubleType)
        ))
    
        // .option("inferSchema",true)
        val frame: DataFrame = spark.read.schema(schema).csv("spark_sql/doc/stu.csv")
        frame.printSchema()
        frame.show()
    
         //.toDF("id","age","name","city","score")
        val df = frame.toDF("id","age","name","city","score")
        df.printSchema()
        df.show()
      }
    }

### 四、有头文件的CSV分析 ###

package blog
    
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    /**
     * @author: 余辉  
     * @blog: https://blog.csdn.net/silentwolfyh
     * @create: 2019-12-29 16:55
     * @description:
     **/
    object DF08_CSV_Header {
    
      Logger.getLogger("org").setLevel(Level.WARN)
    
      def main(args: Array[String]): Unit = {
    
        val spark: SparkSession = SparkSession.builder()
          .appName("DF08_CSV_Header")
          .master("local[*]")
          .getOrCreate()
    
        /** *
         *
         * 有头文件的CSV，程序中在没有设置参数读取头文件，则头文件按列进行展现
         * +---+----+---+---+----+-----+
         * |_c0| _c1|_c2|_c3| _c4|  _c5|
         * +---+----+---+---+----+-----+
         * | id|name|age|sex|city|score|
         *
         * 有头文件的CSV，程序需要设置参数 header 将第一列变成列名称，同时需要设置参数 inferSchema 推断类型
         * 有头文件的CSV，程序需要设置参数 header 以及自定义Schema
         *
         * 如果原始文件中有脏数据，比如:
         * 1、值没有则为null
         * 2、就算少一列，这列的值为null
         *
         * 注意事项：
         * 1、inferSchema 是先从第一行扫描到最后一行，推断数据类型，如果数据量大则耗费大量时间
         * 2、参数设置可以使用options(Map集合)
         */
        val schema = StructType(Array(
          StructField("id", DataTypes.IntegerType),
          StructField("name", DataTypes.StringType),
          StructField("age", DataTypes.IntegerType),
          StructField("city", DataTypes.StringType),
          StructField("score", DataTypes.DoubleType)
        ))
    
        val frame: DataFrame = spark.read.schema(schema).options(Map("header" -> "true"))
          //      .option("header", true).option("inferSchema", true)
          .csv("spark_sql/doc/stu_header.csv")
    
        frame.printSchema()
        frame.show()
      }
    }