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()在RDD前加入处理缺失值
时间: 2024-02-21 21:01:18 浏览: 79
另一个分布式计算框架-Node.js开发
你可以使用 `map` 函数对 RDD 进行处理缺失值,具体做法如下:
1. 首先,将需要处理缺失值的列使用 `filter` 函数筛选出来,比如说,我们要处理的是 `quantity` 列和 `money` 列,那么我们可以这样做:
```scala
val filteredRDD = salesRDD.filter(sale => sale.quantity != 0 && sale.money != 0.0)
```
2. 然后,对筛选出来的 RDD 进行 map 操作,将缺失值替换为默认值,比如说,我们将 `quantity` 列中的缺失值替换为 1,将 `money` 列中的缺失值替换为 0.0,代码如下:
```scala
val cleanedRDD = filteredRDD.map(sale => Sale(sale.cardId, sale.productId, sale.productName, if (sale.quantity == 0) 1 else sale.quantity, if (sale.money == 0.0) 0.0 else sale.money))
```
这样,就可以处理缺失值了。
阅读全文