apache-spark - 在分布式 Spark 系统中嵌入句子的最有效方法是什么？

Question

我有一个包含词嵌入的文件(将词嵌入定义为词的向量表示)，格式如下:

a | [0.23, 0.04, ..., -0.22]
aaron | [0.21, 0.08, ..., -0.41]
... | ...
zebra | [0.97, 0.01, ..., -0.34]

这个文件大约有 2.5 GB。我还有大量的句子想转成向量，例如:

Yes sir, today is a great day.
Would you want to buy that blue shirt?
...
Is there anything else I can help you with?

我的句子嵌入策略现在很简单:

For each sentence:
  For each word:
    Obtain the vector representation of the word using the word embedding file.
  End
  Calculate the average of the word vectors of the sentence.
End

我想既然我有大量的句子要嵌入，我可以使用 Spark 来完成这项任务；将词嵌入作为文件存储在 HDFS 中，并使用 Spark SQL 从 Hive 表中查询句子，但是由于每个节点可能需要访问整个词嵌入文件，这意味着在每个节点中收集整个词嵌入 RDD节点，使得节点之间的通信非常昂贵。

有人对如何有效解决这个问题有任何想法吗？如果问题不明确或者您认为我误解了 Spark 的工作方式，也请告诉我。我仍在学习，非常感谢您的帮助!

提前致谢。

Answer 1

您可以执行以下操作:

1. 将您的词嵌入文件转换为 Spark DataFrame，
   1. 看起来你可以使用像 my_embeddings = spark.read.csv(path="path/to/your_file.csv", sep="|") pyspark api docs 这样的东西

2. 更改 DataFrame 架构 (my_embeddings.schema) 以匹配以下内容:

   1. StructType(List(StructField(word,StringType,true),StructField(vector,ArrayType(FloatType,true),true)))
3. 创建一个小而简单的占位符 Spark ML Word2Vec 模型并保存到 hdfs。 pyspark api docs
   1. 例如model_name.write().overwrite().save("your_hdfs_path_to/model_name")
4. 用上面在 your_hdfs_path_to/model_name/data/ 目录中创建的嵌入 DataFrame 覆盖小型和简单的 Word2Vec 模型数据。
   1. my_embeddings.write.parquet("your_hdfs_path_to/model_name/data/", mode='overwrite')
5. 使用 Word2VecModel.load("your_hdfs_path_to/model_name") 加载 Word2Vec 模型 pyspark api docs
6. 创建一个 Spark DataFrame，其中每个句子都在单独的行中。
7. 使用 RegexTokenizer 标记您的句子 pyspark api docs
8. 使用该模型转换包含标记化句子的 Spark DataFrame。输出列将包含一个与词嵌入向量具有相同维度的向量，它将是句子中所有词向量的平均值。
   1. “Word2VecModel 使用文档中所有单词的平均值将每个文档转换为向量” docs .在您的情况下，“每个文档”将是您的每个句子。 pyspark api docs

一起(猜测某些参数，并使用 pySpark):

import pyspark
from pyspark.sql import SparkSession
from pyspark.ml.feature import RegexTokenizer
from pyspark.ml.feature import Word2Vec, Word2VecModel
from pyspark.ml import Pipeline, PipelineModel


spark = (
    SparkSession
    .builder
    .master('yarn')
    .appName('my_embeddings')
    .getOrCreate()
)

my_embeddings = spark.read.csv(path="path/to/your_embeddings.csv", sep="|")

my_embeddings.schema
# needs to be
# StructType(List(StructField(word,StringType,true),StructField(vector,ArrayType(FloatType,true),true)))

my_sentences = spark.read.csv(path="path/to/your_sentences.csv", sep="|")

tokenizer = (
    RegexTokenizer()
    .setInputCol("sentences")
    .setOutputCol("tokens") 
    .setPattern("\w+")
)

words2vecs = (
    Word2Vec()
    .setInputCol("tokens")
    .setOutputCol("vecs")
    .setMinCount(1)
    .setNumPartitions(5)
    .setStepSize(0.1)
    .setWindowSize(5)
    .setVectorSize(200)
    .setMaxSentenceLength(1)
)


pipeline = (
    Pipeline()
    .setStages([tokenizer, words2vecs])
)

pipe_model = pipeline.fit(my_sentences.limit(100))

pipe_model.stages[1].write().overwrite().save("your_hdfs_path_to/model_name")

my_embeddings.write.parquet("your_hdfs_path_to/model_name/data/", mode='overwrite')

my_embedding_model = Word2VecModel.load("your_hdfs_path_to/model_name")

df_final = my_embedding_model.transform(tokenizer.transform(my_sentences))