Spark：自定义Schema信息将数据集合转换为Spark sql中的DataFrame

清疚 2021-11-05 00:08 436阅读 0赞

# 使用场景 #

在Spark中可以直接读取数据文件  
但是读取到的数据的每一项数据是没有数据类型的  
而且不能使用数据像使用数据表中的字段名那样使用数据  
可以在读取数据的时候对读取到的数据进行设置转换  
设置转换后使用数据可以像使用数据库表中的字段那样  
通过字段名获取数据

# 代码实现 #

import java.util.Properties
    import org.apache.spark.sql.types._
    import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
    
    /**
      * 演示SparkSQL读取各种数据源的数据，进行分析
      */
    object SchemaTsvSpark {
      def main(args: Array[String]): Unit = {
    
        val spark = SparkSession
          .builder()
          .appName("SchemaTsvSpark")
          .master("local[2]")
          //设置sparkSQL中shuffle 时分区数
          .config("spark.sql.shuffle.partitions",2)
          .getOrCreate()
    
        spark.sparkContext.setLogLevel("WARN")
        import spark.implicits._
    
        //自定义Schema信息
        val schema:StructType = StructType(
          Array(
            StructField("user_id",IntegerType,true),
            StructField("item_id",IntegerType,true),
            StructField("rating",DoubleType,true),
            StructField("timestamp",LongType,true)
          )
        )
    
        val mlRatingDF: DataFrame = spark.read
          .option("sep","\t")
          .schema(schema)
          .csv("file:///E:/u.data")
    
    
        mlRatingDF.printSchema()
        mlRatingDF.show(4,false)
        
        spark.read
          .option("sep",",")
          .option("header","true") //获取字段的名称
          .option("inferSchema","true") //自动推断出各列的数据类型
          .csv("file:///E:/ml-100/u_data.csv")
          .printSchema()
         
        spark.stop()
      }
    }