帮我debug:package org.zkpk.lab import org.apache.spark.SparkContext import org.apache.spark.SparkConf import org.apache.spark.rdd.RDD import org.apache.spark.sql.SQLContext object sale_drug { case class Sale(cardId:String,productId:Int,productName:String ,quantity:Int,money:Double) def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("Pharmacy Sales Analysis").setMaster("local") val sc = new SparkContext(conf) sc.setLogLevel("ERROR") val sqlContext = new SQLContext(sc) // load data val data = sc.textFile("/home/zkpk/saledata.txt") // parse data val salesRDD: RDD[Sale] = data.map(line => line.split(" ")).map(arr => Sale(arr(0).toString, arr(1).toInt, arr(2).toString, arr(3).toInt, arr(4).toDouble)) import sqlContext.implicits._ val df = salesRDD.toDF() val nameMoneyDF = df.select("productName","money") nameMoneyDF.show()
时间: 2024-02-21 15:00:47 浏览: 57
在你的代码中,我发现了一个问题:在定义 Sale case class 时,字段 productName 的类型应该是 String 而不是 Int。因为在代码中,该字段的值是从数据文件中读取的一个字符串类型的值。因此,你应该将 Sale case class 的定义修改为:
```
case class Sale(cardId:String,productId:Int,productName:String,quantity:Int,money:Double)
```
另外,如果你的数据文件中使用的是分隔符不是空格,你需要在 `line.split(" ")` 中将空格修改为相应的分隔符。
相关问题
spark.executor.extraJavaOptions=-Dlog4j.logger.org.apache.spark.streaming.kafka.KafkaSource=DEBUG这个配置添加在哪里
这个配置可以添加在 Spark 程序的命令行中,例如在启动程序时使用如下命令:
./bin/spark-submit --conf spark.executor.extraJavaOptions=-Dlog4j.logger.org.apache.spark.streaming.kafka.KafkaSource=DEBUG ...
也可以在 Spark 配置文件中添加,例如 spark-defaults.conf 文件:
spark.executor.extraJavaOptions=-Dlog4j.logger.org.apache.spark.streaming.kafka.KafkaSource=DEBUG
最后,在 Spark 应用程序代码中也可以通过 SparkConf 对象的 setExecutorEnv 方法添加。
阅读全文