I used a pre-trained model from the sentence transformer library to check the similarity between two sentences. Now I need this particular model to be implemented using spark mllib. Any Suggestions? I really appreciate any help you can provide.
1 Answer
One approach that I found to work is to use a Pandas UDF that encodes the text and returns the embedding. This embedding column could then be used with MLlib.
import pandas as pd
import pyspark.sql.functions as F
from pyspark.sql.types import ArrayType, DoubleType, StringType
from sentence_transformers import SentenceTransformer
# import sbert model
model = SentenceTransformer("all-MiniLM-L6-v2")
# sentences to encode
sentences = [
"This framework generates embeddings for each input sentence",
"Sentences are passed as a list of string.",
"The quick brown fox jumps over the lazy dog.",
]
# create spark df with sentences
data = spark.createDataFrame(sentences, StringType(), ["sentences"])
data.show()
# create a pandas udf that will encode the text and return an array of doubles
@F.pandas_udf(returnType=ArrayType(DoubleType()))
def encode(x: pd.Series) -> pd.Series:
return pd.Series(model.encode(x).tolist())
# apply udf and show
data.withColumn("embedding", encode("value")).show()
output
+--------------------+
| value|
+--------------------+
|This framework ge...|
|Sentences are pas...|
|The quick brown f...|
+--------------------+
+--------------------+--------------------+
| value| embedding|
+--------------------+--------------------+
|This framework ge...|[-0.0137173617258...|
|Sentences are pas...|[0.05645250156521...|
|The quick brown f...|[0.04393352568149...|
+--------------------+--------------------+
-
Ok @David... I've done this with a lot more trouble before. My big question for you is this: how do you do this with a GPU and batching? :D We tried on PySpark / Databricks and man, it was hard. I don't know if we ever got it working.– rjurneyCommented Oct 9, 2023 at 17:34
-
For Databricks all I had to do is run the above code on a GPU cluster/runtime. As for batch size there are two "batches". One is how many rows the Pandas UDF processes. which you can set via
spark.sql.execution.arrow.maxRecordsPerBatch
(default 10,000). Then the SBERT library has abatch_size
(default 32) parameter that you can adjust in the.encode()
.– DavidCommented Oct 9, 2023 at 20:37 -