df.select(df.col2 == df.col3).show()把结果添加到新列中

可以使用 `withColumn` 方法来添加结果到新列中，例如：

from pyspark.sql.functions import when

df = df.withColumn("new_col", when(df.col2 == df.col3, "True").otherwise("False"))
df.show()

这将会在 `df` 中添加一个名为 `new_col` 的新列，其中包含 `col2` 等于 `col3` 的行的值为 `True`，否则为 `False`。你可以根据需要修改 `when` 和 `otherwise` 表达式来设置新列的值。

相关问题

df.select(df.col2 == df.col3).show()

这是一个 PySpark DataFrame 的操作，意思是选取 DataFrame 中 col2 和 col3 相等的行并展示出来。具体解释如下：

- df 是一个 PySpark DataFrame 对象；
- select() 方法用于选取满足条件的行；
- df.col2 和 df.col3 分别表示 DataFrame 中的两列；
- == 表示判断两个值是否相等；
- show() 方法用于展示选取后的结果。

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") } }

这是一个使用Spark读取Titanic数据集并对其进行预处理的Scala代码。这个代码将CSV文件读取为一个DataFrame，然后对其中的缺失值进行处理，并计算了生还和死亡人数的统计信息，最后将结果写入CSV文件。

首先，代码创建了一个SparkConf和SparkContext对象，然后创建了一个SparkSession对象，用于读取CSV文件。接下来，代码对数据集中的一些列进行转换，将它们转换为正确的数据类型。然后，代码删除了一些不需要的列。接着，代码计算了每列中缺失值的数量，并将结果存储在一个DataFrame中。接下来，代码定义了一个函数meanAge，用于计算年龄的平均值，并将缺失值替换为该平均值。最后，代码计算了生还和死亡人数的统计信息，并将结果写入CSV文件。

希望这能帮到你！