在 spark scala 中查找平均值会给出空白结果答案

【问题标题】：Finding average value in spark scala gives blank result在 spark scala 中查找平均值会给出空白结果
【发布时间】：2018-05-15 12:31:49
【问题描述】：

我有一个 input.txt 文件。数据如下。

1   1383260400000   0   0.08136262351125882             
1   1383260400000   39  0.14186425470242922 0.1567870050390246  0.16093793691701822 0.052274848528573205    11.028366381681026
1   1383261000000   0   0.13658782275823106         0.02730046487718618 
1   1383261000000   33                  0.026137424264286602
2241    1383324600000   0   0.16869936142032646             
2241    1383324600000   39  0.820500491400199   0.6518011299798726  1.658248219576473   3.4506242774863045  36.71096470849049
2241    1383324600000   49  0.16295028249496815

假设第一列是 id，其他列分别是 col1、col2、col3、col4、col5、col6 和 col7。我想找到每个 id 的 col7 的平均值。基本上我想要我的结果， id，col7 格式的平均值。

这是我迄今为止尝试过的代码。我在 txt 文件中读取了我的数据。然后我创建了一个模式。

val schema = StructType(Seq(
  StructField("ID", IntegerType, true),
  StructField("col1", DoubleType, true),
  StructField("col2", IntegerType, true),
  StructField("col3", DoubleType, true),
  StructField("col4", DoubleType, true),
  StructField("col5", DoubleType, true),
  StructField("col6", DoubleType, true),
  StructField("col7", DoubleType, true)
))

然后我创建了一个数据框。

val data = text.map(line => line.split("\\t")).map(arr => Row.fromSeq(Seq(arr(0).toInt,Try(arr(1).asInstanceOf[DoubleType]) getOrElse(0.0),Try(arr(2).toInt) getOrElse(0),Try(arr(3).toDouble) getOrElse(0.0),Try(arr(4).toDouble) getOrElse(0.0),Try(arr(5).toDouble) getOrElse(0.0),Try(arr(6).toDouble) getOrElse(0.0),Try(arr(7).asInstanceOf[DoubleType]) getOrElse(0.0))))

最后保存为txt文件。

val res1 = df.groupBy("ID").agg(avg("col7"))

res1.rdd.saveAsTextFile("/stuaverage/spoutput12")

当我运行它时，我会得到几个结果为空白的文件。例如

[1068,0.0]
[1198,0.0]
[1344,0.0]
[1404,0.0]
[1537,0.0]
[1675,0.0]
[1924,0.0]
[193,0.0]
[211,0.0]
[2200,0.0]
[2225,0.0]
[2663,0.0]
[2888,0.0]
[3152,0.0]
[3235,0.0]

第一列是正确的。但是对于第二列，我应该得到一个值。（尽管某些行缺少值）

请帮忙。

【问题讨论】：

数据格式有问题。有时有空格，有时有制表或双制表...所以实际发生的情况如下：java.lang.NumberFormatException: For input string: "1 1383260400000 0 0.08136262351125882 " 但由于您使用的是 Try().getOrElse，因此您的所有值都是 @ 987654327@ 但您无法捕捉到错误。您需要提供一个更正的数据格式，其中实际上只有一个表格可以分开。
@RameshMaharjan 我不需要。
好的 :) 因为您昨天以消极的想法评论了答案。我以为你投了反对票。对不起我的误解。但我很惊讶有人在不理解答案的情况下投了反对票。
我昨天写信给你，我认为你的答案不正确，但 OP 有我在我的 cmets 中解释的其他问题。这似乎对他有用。我的cmets仍然站立。你已经有一段时间了。我不批评为负面评论，但棘手的是粗鲁......我不喜欢粗鲁的人。

标签： scala apache-spark apache-spark-sql

【解决方案1】：

试试这个更简洁的版本（假设您使用 spark-shell 工作）。它应该可以工作。

val df = spark
  .read
  .option("header","false")
  .option("sep","\t")
  .option("inferSchema","true")
  .csv("...input...")
  .toDF("ID","col1","col2","col3","col4","col5","col6","col7")

val result = df.groupBy("ID").mean("col7")

result
  .write
  .option("header","true")
  .option("sep",";")
  .csv("...output...")

【讨论】：

Konstantin Kotochigov：您能解释一下这将如何计算 col7 的平均值吗？我想要每个 ID，col7 平均值。
我错过了代码中的一个参数，当然应该是 mean("col7") 。 groupBy("ID") 将根据每个 ID 对记录进行分组， mean("col7") 将计算列的平均值。
问题是我在问题下的评论中所说的。

【解决方案2】：

我建议你使用 sqlContext api 并使用你定义的模式

val df = sqlContext.read
  .format("com.databricks.spark.csv")
  .option("delimiter", "\\t")
  .schema(schema)
  .load("path to your text file")

架构是

val schema = StructType(Seq(
  StructField("ID", IntegerType, true),
  StructField("col1", DoubleType, true),
  StructField("col2", IntegerType, true),
  StructField("col3", DoubleType, true),
  StructField("col4", DoubleType, true),
  StructField("col5", DoubleType, true),
  StructField("col6", DoubleType, true),
  StructField("col7", DoubleType, true)
))

之后，您只需在分组的dataframe 上应用avg 函数

import org.apache.spark.sql.functions._
val res1 = df.groupBy("ID").agg(avg("col1"),avg("col2"),avg("col3"),avg("col4"),avg("col5"),avg("col6"),avg("col7"))

终于可以从dataframe直接保存到csv。你不需要转换成rdd

  res1.coalesce(1).write.csv("/stuaverage/spoutput12")

【讨论】：

【解决方案3】：

问题是您以错误的方式转换col7，您尝试将其转换为DoubleType，而不是将其解析为scalaDouble（使用.toDouble）。您的演员将始终抛出异常，因此col7 将始终为 0.0。这有效：

val rdd = sqlContext.textFile("input.txt")
  .map(line => line.split("\\t"))
    .map((arr: Array[String]) => Row(
    arr(0).toInt,
    Try(arr(1).toDouble) getOrElse (0.0),
    Try(arr(2).toInt) getOrElse (0),
    Try(arr(3).toDouble) getOrElse (0.0),
    Try(arr(4).toDouble) getOrElse (0.0),
    Try(arr(5).toDouble) getOrElse (0.0),
    Try(arr(6).toDouble) getOrElse (0.0),
    Try(arr(7).toDouble) getOrElse (0.0)
    )
  )

【讨论】：