代码之家 › 专栏 › 技术社区 › Daniil Andreyevich Baunov

将两个Seq列之间的相关性计算为第三列的正确方法

correlation apache-spark scala

Daniil Andreyevich Baunov · 技术社区 · 7 年前

ID:Long, ratings1:Seq[Double], ratings2:Seq[Double]

对于每一行,我需要计算这些向量之间的相关性。

我提出了以下解决方案,由于我必须为每个Seq创建RDD,该解决方案似乎效率低下(不像Jarrod Roberson提到的那样工作):

val similarities = ratingPairs.map(row => {
      val ratings1 = sc.parallelize(row.getAs[Seq[Double]]("ratings1"))
      val ratings2 = sc.parallelize(row.getAs[Seq[Double]]("ratings2"))
      val corr:Double = Statistics.corr(ratings1, ratings2)

      Similarity(row.getAs[Long]("ID"), corr)
    })

有没有办法正确计算这种相关性?

1 回复 | 直到 7 年前

Tzach Zohar 7 年前

假设有一个数组的相关函数:

def correlation(arr1: Array[Double], arr2: Array[Double]): Double

(对于完全独立于Spark的该功能的潜在实现,您可以提出单独的问题或在线搜索,有一些足够接近的资源,例如:。 this implementation ).

现在,剩下要做的就是用一个并使用它:

import org.apache.spark.sql.functions._
import spark.implicits._

val corrUdf = udf {
  (arr1: Seq[Double], arr2: Seq[Double]) => correlation(arr1.toArray, arr2.toArray)
}

val result = df.select($"ID", corrUdf($"ratings1", $"ratings2") as "correlation")

推荐文章

Geoffrey · Pyspark:将数据帧保存到多个具有单个文件特定大小的镶木地板文件中

1 年前

Bruno Peixoto · Spark群集CI管道构建失败

1 年前

codebot · 将df从pandas转换为PySpark时会删除列名

1 年前

mcsilvio · 在foreach中组织联接的最佳方式是什么?

1 年前

Dhruv · 在sbt控制台上运行Spark

1 年前

Leonard · Pyspark:JSON到Pyspark数据帧

1 年前

billie class · 将列中的值重写为列表中的下一个值

2 年前

Calcutta · Google Colab中的Spark SQL在大数据上失败

2 年前

Doraemon · PySpark:使用不同值的字符串类型列创建聚合列

2 年前

OdiumPura · 使用JDBC(Sql server)查询tempview

2 年前