object Data_aggregation { def main(args: Array[String]): Unit = { //这里更换其它类型的蔬菜数据 val inputFile = "datas/ziganlan.csv"; val conf = new SparkConf().setAppName("demo1").setMaster("local") val sc = new SparkContext(conf) val sqc = new SQLContext(sc) // 创建spark session val spark = SparkSession.builder().config(conf).getOrCreate() val textFile = sc.textFile(inputFile) val FileRDD = textFile.map( line => { val year = line.split(",")(4).split("-")(0) val month = line.split(",")(4).split("-")(1) val times = year+"-"+month val name = line.split(",")(0) val lowprice = line.split(",")(1).toDouble val minprice = line.split(",")(2).toDouble val topprice = line.split(",")(3).toDouble (name,lowprice,minprice,topprice,times) } ) FileRDD.foreach(println)

这段代码是使用Spark进行数据聚合的代码。首先，指定了输入文件路径和Spark的配置信息，创建了SparkContext和SQLContext。接着，使用textFile()方法读取输入文件中的每一行数据，并使用map()方法对每一行数据进行操作，将其转化为一个元组。其中，元组包含了蔬菜名称、最低价、平均价、最高价和时间信息。通过对每一行数据的处理，可以将数据按照时间和蔬菜名称进行聚合。最后，使用foreach()方法将聚合后的数据打印出来。这段代码是数据处理中数据聚合的基础代码，用于对大规模数据进行聚合分析。