使用pyspark/pandas跨行查找常用詞

我有一個帶有管道分隔符的文本檔案，如下所示

person_id | category | notes
1|A|He bought cat
1|A|He bought dog
1|B|He has hen
2|A|Switzerland Australia
2|A|Australia

我想按 person_id 和 category 分組，只找到所有行中重復的單詞

預期產出

1|A|He bought
1|B|he has hen
2|A|Australia

我已經按 person_id 和 category 購買了每個 using group 的字數，但我無法獲得輸出

我使用 rdd word count 和 spark-sql 獲得了使用 group by 的字數，如下所示

person_id | category | notes
1|A|He (2) bought(2) cat(1) dog(1)
1|B|He(1) has(1) hen(1)
2|A|Switzerland(1) Australia(2)

uj5u.com熱心網友回復：

您可以使用 Spark 陣列函式來實作：

1. 拆分列notes以獲取單詞陣列
2. group byperson_id并category收集單詞串列
3. 通過使用高階函式檢查它是否存在于所有收集的子陣列（即單詞）中來過濾結果陣列 filter

import pyspark.sql.functions as F

df1 = df.withColumn("notes", F.split("notes", " ")) \
    .groupBy("person_id", "category") \
    .agg(F.collect_list(F.col("notes")).alias("notes")) \
    .withColumn("w", F.array_distinct(F.flatten("notes"))) \
    .withColumn("notes", F.array_join(F.expr("filter(w, x -> size(filter(notes, y -> array_contains(y, x))) = size(notes))"), " ")) \
    .drop("w")

df1.show()
# --------- -------- ---------- 
#|person_id|category|notes     |
# --------- -------- ---------- 
#|1        |A       |He bought |
#|1        |B       |He has hen|
#|2        |A       |Australia |
# --------- -------- ---------- 

uj5u.com熱心網友回復：

您可以將notes字串拆分為單詞，然后分解單詞。最后count是notes每個的數量person_id, category和單詞 per 的出現次數person_id, category。如果它們的計數相等，則通過 構造單詞collect_list。

from pyspark.sql import functions as F
from pyspark.sql import Window

data = [(1, "A", "He bought cat"),
(1, "A", "He bought dog"),
(1, "B", "He has hen"),
(2, "A", "Switzerland Australia"),
(2, "A", "Australia"),]

df = spark.createDataFrame(data, ("person_id", "category", "notes", ))

window_spec_category = Window.partitionBy("person_id", "category")

df_word = df.withColumn("category_count", F.count("*").over(window_spec_category))\
            .select("person_id", "category", F.posexplode(F.split(F.col("notes"), " ")).alias("pos", "word"))

window_spec_word = Window.partitionBy("person_id", "category", "word")

matching_words = df_word.withColumn("word_count", F.count("*").over(window_spec_word))\
        .withColumn("rn", F.row_number().over(window_spec_word.orderBy(F.lit(None))))\
        .filter(F.col("word_count") == F.col("category_count"))\
        .filter(F.col("rn") == F.lit(1))\
        .drop("rn")

window_spec_collect = window_spec_category.orderBy("pos").rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

matching_words.withColumn("result", F.concat_ws(" ", F.collect_list("word").over(window_spec_collect)))\
              .withColumn("rn", F.row_number().over(window_spec_category.orderBy(F.lit(None))))\
              .filter(F.col("rn") == F.lit(1))\
              .select("person_id", "category", "result")\
              .show()

輸出

 --------- -------- ---------- 
|person_id|category|    result|
 --------- -------- ---------- 
|        1|       A| He bought|
|        1|       B|He has hen|
|        2|       A| Australia|
 --------- -------- ---------- 

uj5u.com熱心網友回復：

另一種選擇是定義一個 UDF 來查找字串陣列的交集：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.getOrCreate()
data = [
    {"person_id": 1, "category": "A", "notes": "He bought cat"},
    {"person_id": 1, "category": "A", "notes": "He bought dog"},
    {"person_id": 1, "category": "B", "notes": "He has hen"},
    {"person_id": 2, "category": "A", "notes": "Switzerland Australia"},
    {"person_id": 2, "category": "A", "notes": "Australia"},
]


def common(x):
    l = [i.split() for i in x]
    return " ".join(sorted(set.intersection(*map(set, l)), key=l[0].index))

df = spark.createDataFrame(data)
df = df.groupBy(["person_id", "category"]).agg(F.collect_list("notes").alias("b"))
df = df.withColumn("result", F.udf(common)(F.col("b")))

結果：

 --------- -------- ----------                                                  
|person_id|category|result    |
 --------- -------- ---------- 
|1        |A       |He bought |
|1        |B       |He has hen|
|2        |A       |Australia |
 --------- -------- ---------- 

標籤：熊猫 阿帕奇火花 火花 apache-spark-sql

上一篇：使用Scala更新Map[String,Map]中的值

下一篇：Spark-shell不匯入指定的jar檔案