Die Berechnung der Kosinus-ähnlichkeit zwischen allen Zeilen des dataframe in pyspark

Können Sie die mllib - Paket zum berechnen der L2 norm des TF-IDF jeder Zeile. Dann multiplizieren Sie die Tabelle mit sich selbst, um die Kosinus-ähnlichkeit als das Skalarprodukt von zwei durch zwei L2Normen:

1. RDD

rdd = sc.parallelize([[1, "Delhi, Mumbai, Gandhinagar"],[2, " Delhi, Mandi"], [3, "Hyderbad, Jaipur"]])

• Berechnen TF-IDF:

  documents = rdd.map(lambda l: l[1].replace(" ", "").split(","))
  
  from pyspark.mllib.feature import HashingTF, IDF
  hashingTF = HashingTF()
  tf = hashingTF.transform(documents)

Können Sie die Anzahl der features in HashingTF um die Funktion matrix kleiner (weniger Spalten).

    tf.cache()
    idf = IDF().fit(tf)
    tfidf = idf.transform(tf)

• Berechnen L2norm:

  from pyspark.mllib.feature import Normalizer
  labels = rdd.map(lambda l: l[0])
  features = tfidf
  
  normalizer = Normalizer()
  data = labels.zip(normalizer.transform(features))

• Berechnen Kosinus-ähnlichkeit durch die Multiplikation der matrix mit sich selbst:

  from pyspark.mllib.linalg.distributed import IndexedRowMatrix
  mat = IndexedRowMatrix(data).toBlockMatrix()
  dot = mat.multiply(mat.transpose())
  dot.toLocalMatrix().toArray()
  
      array([[ 0.        ,  0.        ,  0.        ,  0.        ],
             [ 0.        ,  1.        ,  0.10794634,  0.        ],
             [ 0.        ,  0.10794634,  1.        ,  0.        ],
             [ 0.        ,  0.        ,  0.        ,  1.        ]])

  ODER: Mit einem kartesischen Produkt und die Funktion dot auf numpy arrays:

  data.cartesian(data)\
      .map(lambda l: ((l[0][0], l[1][0]), l[0][1].dot(l[1][1])))\
      .sortByKey()\
      .collect()
  
      [((1, 1), 1.0),
       ((1, 2), 0.10794633570596117),
       ((1, 3), 0.0),
       ((2, 1), 0.10794633570596117),
       ((2, 2), 1.0),
       ((2, 3), 0.0),
       ((3, 1), 0.0),
       ((3, 2), 0.0),
       ((3, 3), 1.0)]

2. DataFrame

Da Sie zu sein scheinen mit dataframes, die Sie verwenden können, die spark mlPaket statt:

import pyspark.sql.functions as psf
df = rdd.toDF(["ID", "Office_Loc"])\
    .withColumn("Office_Loc", psf.split(psf.regexp_replace("Office_Loc", " ", ""), ','))

• Berechnen Sie TF-IDF:

  from pyspark.ml.feature import HashingTF, IDF
  hashingTF = HashingTF(inputCol="Office_Loc", outputCol="tf")
  tf = hashingTF.transform(df)
  
  idf = IDF(inputCol="tf", outputCol="feature").fit(tf)
  tfidf = idf.transform(tf)

• Berechnen L2 norm:

  from pyspark.ml.feature import Normalizer
  normalizer = Normalizer(inputCol="feature", outputCol="norm")
  data = normalizer.transform(tfidf)

• Berechnen Sie matrix-Produkt:

  from pyspark.mllib.linalg.distributed import IndexedRow, IndexedRowMatrix
  mat = IndexedRowMatrix(
      data.select("ID", "norm")\
          .rdd.map(lambda row: IndexedRow(row.ID, row.norm.toArray()))).toBlockMatrix()
  dot = mat.multiply(mat.transpose())
  dot.toLocalMatrix().toArray()

  ODER: mit einem join und einer UDF für die Funktion dot:

  dot_udf = psf.udf(lambda x,y: float(x.dot(y)), DoubleType())
  data.alias("i").join(data.alias("j"), psf.col("i.ID") < psf.col("j.ID"))\
      .select(
          psf.col("i.ID").alias("i"), 
          psf.col("j.ID").alias("j"), 
          dot_udf("i.norm", "j.norm").alias("dot"))\
      .sort("i", "j")\
      .show()
  
      +---+---+-------------------+
      |  i|  j|                dot|
      +---+---+-------------------+
      |  1|  2|0.10794633570596117|
      |  1|  3|                0.0|
      |  2|  3|                0.0|
      +---+---+-------------------+

Diesem tutorial listet die verschiedenen Methoden zu multiplizieren großen Maßstab Matrizen: https://labs.yodas.com/large-scale-matrix-multiplication-with-pyspark-or-how-to-match-two-large-datasets-of-company-1be4b1b2871e