Efficient string matching in Apache Spark

by

I wouldn’t use Spark in the first place, but if you are really committed to the particular stack, you can combine a bunch of ml transformers to get best matches. You’ll need Tokenizer (or split):

import org.apache.spark.ml.feature.RegexTokenizer

val tokenizer = new RegexTokenizer().setPattern("").setInputCol("text").setMinTokenLength(1).setOutputCol("tokens")

NGram (for example 3-gram)

import org.apache.spark.ml.feature.NGram

val ngram = new NGram().setN(3).setInputCol("tokens").setOutputCol("ngrams")

Vectorizer (for example CountVectorizer or HashingTF):

import org.apache.spark.ml.feature.HashingTF

val vectorizer = new HashingTF().setInputCol("ngrams").setOutputCol("vectors")

and LSH:

import org.apache.spark.ml.feature.{MinHashLSH, MinHashLSHModel}

// Increase numHashTables in practice.
val lsh = new MinHashLSH().setInputCol("vectors").setOutputCol("lsh")

Combine with Pipeline

import org.apache.spark.ml.Pipeline

val pipeline = new Pipeline().setStages(Array(tokenizer, ngram, vectorizer, lsh))

Fit on example data:

val query = Seq("Hello there 7l | real|y like Spark!").toDF("text")
val db = Seq(
  "Hello there 😊! I really like Spark ❤️!", 
  "Can anyone suggest an efficient algorithm"
).toDF("text")

val model = pipeline.fit(db)

Transform both:

val dbHashed = model.transform(db)
val queryHashed = model.transform(query)

and join

model.stages.last.asInstanceOf[MinHashLSHModel]
  .approxSimilarityJoin(dbHashed, queryHashed, 0.75).show

+--------------------+--------------------+------------------+                  
|            datasetA|            datasetB|           distCol|
+--------------------+--------------------+------------------+
|[Hello there 😊! ...|[Hello there 7l |...|0.5106382978723405|
+--------------------+--------------------+------------------+

The same approach can be used in Pyspark

from pyspark.ml import Pipeline
from pyspark.ml.feature import RegexTokenizer, NGram, HashingTF, MinHashLSH

query = spark.createDataFrame(
    ["Hello there 7l | real|y like Spark!"], "string"
).toDF("text")

db = spark.createDataFrame([
    "Hello there 😊! I really like Spark ❤️!", 
    "Can anyone suggest an efficient algorithm"
], "string").toDF("text")


model = Pipeline(stages=[
    RegexTokenizer(
        pattern="", inputCol="text", outputCol="tokens", minTokenLength=1
    ),
    NGram(n=3, inputCol="tokens", outputCol="ngrams"),
    HashingTF(inputCol="ngrams", outputCol="vectors"),
    MinHashLSH(inputCol="vectors", outputCol="lsh")
]).fit(db)

db_hashed = model.transform(db)
query_hashed = model.transform(query)

model.stages[-1].approxSimilarityJoin(db_hashed, query_hashed, 0.75).show()
# +--------------------+--------------------+------------------+
# |            datasetA|            datasetB|           distCol|
# +--------------------+--------------------+------------------+
# |[Hello there 😊! ...|[Hello there 7l |...|0.5106382978723405|
# +--------------------+--------------------+------------------+

Related

More Related Contents:

  1. How to add a constant column in a Spark DataFrame?
  2. Spark Dataframe distinguish columns with duplicated name
  3. How to change dataframe column names in pyspark?
  4. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  5. Explode in PySpark
  6. Python worker failed to connect back
  7. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  8. Spark RDD to DataFrame python
  9. Filtering a Pyspark DataFrame with SQL-like IN clause
  10. Spark union of multiple RDDs
  11. Dividing complex rows of dataframe to simple rows in Pyspark
  12. Spark DataFrame TimestampType – how to get Year, Month, Day values from field?
  13. Why is Apache-Spark – Python so slow locally as compared to pandas?
  14. Spark add new column to dataframe with value from previous row
  15. Tuning parameters for implicit pyspark.ml ALS matrix factorization model through pyspark.ml CrossValidator
  16. How to create a custom Estimator in PySpark
  17. Pyspark: Replacing value in a column by searching a dictionary
  18. Pivot String column on Pyspark Dataframe
  19. PySpark DataFrames – way to enumerate without converting to Pandas?
  20. Reduce a key-value pair into a key-list pair with Apache Spark
  21. Using UDF ignores condition in when
  22. How do I set the driver’s python version in spark?
  23. PySpark in iPython notebook raises Py4JJavaError when using count() and first()
  24. pyspark parse fixed width text file
  25. Concatenating string by rows in pyspark
  26. PySpark: match the values of a DataFrame column against another DataFrame column
  27. What is the equivalent to scala.util.Try in pyspark?
  28. When to cache a DataFrame?
  29. environment variables PYSPARK_PYTHON and PYSPARK_DRIVER_PYTHON
  30. What is the difference between spark-submit and pyspark?

Leave a Comment