代码如下:

import org.apache.spark.SparkConf  

import org.apache.spark.sql.{DataFrame, SparkSession}  

import org.apache.spark.sql.functions._

import org.apache.spark.sql.types.DoubleType

 

object turbine_analysis03 {  

  def main(args: Array[String]): Unit = {  

    // 创建SparkSQL的运行环境  

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Task02")  

    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    

    // 文件路径  

    val path = "/data/bigfiles/b18905c9-028a-4475-9c57-a35c8323d484.csv"

    

    /********** Begin **********/  

    // 读取CSV文件并转换WindSpeed列为Double类型

    val df = spark.read.format("csv").option("header", "true").load(path)

    val df1: DataFrame = df.withColumn("WindSpeed", col("WindSpeed").cast(DoubleType))  

    

    // 创建临时视图以便SQL查询

    df1.createOrReplaceTempView("turbine")    

    

    // 使用SQL查询进行风力等级分析

    spark.sql(  

      """  

        | select  

        | level,  

        | count(*) frequency  

        | from (  

        | select  

        | WindSpeed,  

        | case when WindSpeed>=0.0 and WindSpeed<=0.2 then '无风'  

        | when WindSpeed>=0.3 and WindSpeed<=1.5 then '软风'  

        | when WindSpeed>=1.6 and WindSpeed<=3.3 then '轻风'  

        | when WindSpeed>=3.4 and WindSpeed<=5.4 then '微风'  

        | when WindSpeed>=5.5 and WindSpeed<=7.9 then '和风'  

        | when WindSpeed>=8.0 and WindSpeed<=10.7 then '劲风'  

        | when WindSpeed>=10.8 and WindSpeed<=13.8 then '强风'  

        | when WindSpeed>=13.9 and WindSpeed<=17.1 then '疾风'  

        | when WindSpeed>=17.2 and WindSpeed<=20.7 then '大风'  

        | when WindSpeed>=20.8 and WindSpeed<=24.4 then '烈风'  

        | when WindSpeed>=24.5 and WindSpeed<=28.4 then '狂风'  

        | when WindSpeed>=28.5 and WindSpeed<=32.6 then '暴风'  

        | else '台风' end level  

        | from turbine )  

        | group by level  

        | order by frequency desc  

        |""".stripMargin)  

      .repartition(1) // 合并结果到一个分区

      .write.format("jdbc") // 写入JDBC数据源

      .option("driver","com.mysql.jdbc.Driver")    

      .option("url","jdbc:mysql://127.0.0.1:3306/turbine?useUnicode=true&characterEncoding=utf-8")    

      .option("dbtable","wind_info")    

      .option("user","root")    

      .option("password","123123")    

      .mode("append") // 追加模式保存数据

      .save()

    /********** End **********/  

    spark.close()  

  }  

}