import breeze.numerics.round import breeze.stats.mean import org.apache.spark.sql.functions.col import org.apache.spark.sql.types.{DoubleType, IntegerType} import org.apache.spark.{SparkConf, SparkContext} import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.DataFrame object Titanic_c { def main(args: Array[String]) = { Logger.getLogger("org").setLevel(Level.ERROR) val conf = new SparkConf().setAppName("Titanic_c").setMaster("local[2]") val sc = new SparkContext(conf) val spark = org.apache.spark.sql.SparkSession.builder .master("local") .appName("Titanic") .getOrCreate; val df = spark.read .format("csv") .option("header", "true") .option("mode", "DROPMALFORMED") .load("datasets/Titanic_s.csv") import spark.implicits._ df.withColumn("Pclass", df("Pclass").cast(IntegerType)) .withColumn("Survived", df("Survived").cast(IntegerType)) .withColumn("Age", df("Age").cast(DoubleType)) .withColumn("SibSp", df("SibSp").cast(IntegerType)) .withColumn("Parch", df("Parch").cast(IntegerType)) .withColumn("Fare", df("Fare").cast(DoubleType)) val df1 = df.drop("PassengerId").drop("Name").drop("Ticket").drop("Cabin") val columns = df1.columns val missing_cnt = columns.map(x => df1.select(col(x)).where(col(x).isNull).count) val result_cnt = sc.parallelize(missing_cnt.zip(columns)).toDF("missing_cnt", "column_name") result_cnt.show() def meanAge(dataFrame: DataFrame): Double = { dataFrame .select("Age") .na.drop() .agg(round(mean("Age"), )) .first() .getDouble(0) } val df2 = df1 .na.fill(Map( "Age" -> meanAge(df1), "Embarked" -> "S")) val survived_count = df2.groupBy("Survived").count() survived_count.show() survived_count.coalesce(1).write.option("header", "true").csv("datasets/survived_count.csv") } }
时间: 2023-11-15 21:06:29 浏览: 198
这是一个使用Spark读取Titanic数据集并对其进行预处理的Scala代码。这个代码将CSV文件读取为一个DataFrame,然后对其中的缺失值进行处理,并计算了生还和死亡人数的统计信息,最后将结果写入CSV文件。
首先,代码创建了一个SparkConf和SparkContext对象,然后创建了一个SparkSession对象,用于读取CSV文件。接下来,代码对数据集中的一些列进行转换,将它们转换为正确的数据类型。然后,代码删除了一些不需要的列。接着,代码计算了每列中缺失值的数量,并将结果存储在一个DataFrame中。接下来,代码定义了一个函数meanAge,用于计算年龄的平均值,并将缺失值替换为该平均值。最后,代码计算了生还和死亡人数的统计信息,并将结果写入CSV文件。
希望这能帮到你!
相关问题
代码如下: import breeze.numerics.round import org.apache.spark.sql.functions.col import org.apache.spark.sql.types.{DoubleType, IntegerType} import org.apache.spark.{SparkConf, SparkContext} import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.DataFrame object Titanic_c { def main(args: Array[String]) = { Logger.getLogger("org").setLevel(Level.ERROR) val conf = new SparkConf().setAppName("Titanic_c").setMaster("local[2]") val sc = new SparkContext(conf) val spark = org.apache.spark.sql.SparkSession.builder .master("local") .appName("Titanic") .getOrCreate; val df = spark.read .format("csv") .option("header", "true") .option("mode", "DROPMALFORMED") .load("datasets/Titanic_s.csv") import spark.implicits._ df.withColumn("Pclass", df("Pclass").cast(IntegerType)) .withColumn("Survived", df("Survived").cast(IntegerType)) .withColumn("Age", df("Age").cast(DoubleType)) .withColumn("SibSp", df("SibSp").cast(IntegerType)) .withColumn("Parch", df("Parch").cast(IntegerType)) .withColumn("Fare", df("Fare").cast(DoubleType)) val df1 = df.drop("PassengerId").drop("Name").drop("Ticket").drop("Cabin") val columns = df1.columns val missing_cnt = columns.map(x => df1.select(col(x)).where(col(x).isNull).count) val result_cnt = sc.parallelize(missing_cnt.zip(columns)).toDF("missing_cnt", "column_name") result_cnt.show() import breeze.stats._ def meanAge(dataFrame: DataFrame): Double = { dataFrame .select("Age") .na.drop() .agg(round(mean("Age"), 0)) .first() .getDouble(0) } val df2 = df1 .na.fill(Map( "Age" -> meanAge(df1), "Embarked" -> "S")) val survived_count = df2.groupBy("Survived").count() survived_count.show() survived_count.coalesce(1).write.option("header", "true").csv("datasets/survived_count.csv") } }
我发现问题可能出在 `meanAge` 函数中的 `round(mean("Age"), 0)` 这句代码上。这个代码中的 `mean` 函数是 Spark SQL 中的函数,但是在 `breeze.stats` 包中也有一个名为 `mean` 的函数,这可能导致了混淆。为了解决这个问题,你可以尝试将 `mean` 函数改名为其他的变量名,例如:
```scala
def meanAge(dataFrame: DataFrame): Double = {
val ageMean = dataFrame
.select("Age")
.na.drop()
.agg(round(mean("Age"), 0))
.first()
.getDouble(0)
ageMean
}
```
这样就将 `mean` 函数改名为了 `ageMean` 变量,并避免了与 `breeze.stats` 包中的函数重名的问题。希望对你有帮助!
import breeze.linalg._pom.xml文件中需要调取哪个依赖
`import breeze.linalg._` 这行代码通常是在Scala项目中导入Breeze库的矩阵和向量运算相关的部分。Breeze是一个用于数值计算的 Scala 库,它提供了高性能的科学计算功能。要在`.pom.xml` 文件中添加对 Breeze 的依赖,你需要在`<dependencies>`标签内添加类似这样的内容:
```xml
<dependency>
<groupId>org.scalanlp</groupId>
<artifactId>breeze_2.11</artifactId>
<version>0.14.0-M5</version> <!-- 请替换为你实际使用的版本 -->
</dependency>
```
这里,`groupId`通常是组织ID,`artifactId`对应于项目的名称,`breeze_2.11`表示这个库是针对 Scala 2.11编写的。请确保将`version`替换为你想要使用的Breeze库的实际版本。
如果你正在处理的是Maven项目,上述信息应该会帮助你在构建时正确地引用Breeze库。如果使用Sbt(Scala Build Tool),则会在build.sbt文件中配置依赖。
阅读全文
相关推荐












