Заменить / посмотреть значения одного столбца в spark df значениями в другом pyspark фрейма данных

Question

У меня есть один фрейм данных со списком токенов.

data1 = [(1,  ["This","is", "category", "A"]),
    (2,  ["This", "is", "category", "B","This", "is", "category", "B"]),
    (3,  ["This", "is", "category", "F","This", "is", "category", "C"])]

df2=spark.createDataFrame(data1).withColumnRenamed('_1','category').withColumnRenamed('_2','tokens')

У меня есть еще один фрейм данных с токенами и их векторным представлением. Вот схема для второго

StructType(List(StructField(word,StringType,true),StructField(vector,ArrayType(DoubleType,true),true)))

word  vector
you   [0.04986, 0.5678]

Я хочу посмотреть список токенов во фрейме данных с векторным представлением и вычислить среднее значение в pyspark.

Пожалуйста, дайте мне знать, как Я могу сделать это эффективно в pyspark.

Лог c в python / panda соответствует http://nadbordrozd.github.io/blog/2016/05/20/text-classification-with-word2vec/

return np.array([
            np.mean([word2vec[w] for w in words if w in word2vec]
                    or [np.zeros(dim)], axis=0)
            for words in X

Answer 1

Для этого может быть несколько способов. Я могу рассказать вам о способе, который я обычно предпочел бы сделать ..

Шаг # 1: преобразовать 2-й кадр данных в карту и передать объект. Предоставляя пример в scala

val map = df.collect().map(r => r.getString(0) -> r.getString(1)) 
val broadcasted_map = sc.broadcast(map)

Шаг # 2: Создайте функцию UDF для выполнения требуемой средней операции, а внутри функции UDF вы можете получить доступ к широковещательной карте, используя broadcasted_map.values.get ("word")

val udf_function = udf((token: String) => {
// place your mean function here and access map here for word lookup
})
df.withcolumn("mean_value", udf_function(col("tokens"))) //add your transformation column

Поскольку карта Вещание будет доступно во всех рабочих узлах, и поиск будет локальным по отношению к узлу и не требует перемешивания.