他の2つのエントリを比較して新しいPySpark DataFrameを生成するには？

Question

文字列フィールドを含むPyspark DataFrameを検索し、それぞれに表示されるキーワード文字列を決定したいと考えています。

+-----------+----------+ 
|  city|  state| 
+-----------+----------+ 
| Seattle|Washington| 
|Los Angeles|California| 
+-----------+----------+ 

DataFrameこの中で私が検索したい：私はタイプのキーワードがで表示されますを識別する新しいDataFrameを作成したい

+----------------------------------------+------+ 
|body         |source| 
+----------------------------------------+------+ 
|Seattle is in Washington.    |a  | 
|Los Angeles is in California   |b  | 
|Banana is a fruit      |c  | 
|Seattle is not in New Hampshire   |d  | 
|California is home to Los Angeles  |e  | 
|Seattle, California is not a real place.|f  | 
+----------------------------------------+------+ 

私は、キーワードのDataFrameを以下持っていると言います各ソース。したがって、最終結果は

+-----------+------+-----+ 
|name  |source|type | 
+-----------+------+-----+ 
|Seattle |a  |city | 
|Washington |a  |state| 
|Los Angeles|b  |city | 
|California |b  |state| 
|Seattle |d  |city | 
|Los Angeles|e  |city | 
|California |e  |state| 
|Seattle |f  |city | 
|California |f  |state| 
+-----------+------+-----+ 

となります。この結果はどのように取得できますか？ joinを使用してこれらのキーワードを含む文字列を隔離することはできますが、どの特定のキーワードが一致したかをトラッキングしてその情報を使用して新しい列を作成する方法はわかりません。

Answer 1

まずは、データフレームを作成し、変更してみましょう：

import pyspark.sql.functions as psf 
keywords_df = sc.parallelize([["Seattle", "Washington"], ["Los Angeles", "California"]])\ 
    .toDF(["city", "state"]) 
keywords_df = keywords_df\ 
    .withColumn("struct", psf.explode(psf.array(
     psf.struct(psf.col("city").alias("word"), psf.lit("city").alias("type")), 
     psf.struct(psf.col("state").alias("word"), psf.lit("state").alias("type")) 
    )))\ 
    .select("struct.*") 
keywords_df.show() 

    +-----------+-----+ 
    |  word| type| 
    +-----------+-----+ 
    | Seattle| city| 
    | Washington|state| 
    |Los Angeles| city| 
    | California|state| 
    +-----------+-----+ 

あなたのキーワードはスペースが含まれていなかった場合は、あなただけの1を取得するexplodedを持っているだろうと、言葉にsplitあなたの文章を持つことができます各行の単語。その後、キーワードデータフレームをjoinにすることができました。 Los Angelesのため、ここには該当しません。

res = text_df.join(keywords_df, text_df.body.contains(keywords_df.word)).drop("body") 
res.show() 

    +------+-----------+-----+ 
    |source|  word| type| 
    +------+-----------+-----+ 
    |  a| Seattle| city| 
    |  a| Washington|state| 
    |  b|Los Angeles| city| 
    |  b| California|state| 
    |  d| Seattle| city| 
    |  f| Seattle| city| 
    |  e|Los Angeles| city| 
    |  e| California|state| 
    |  f| California|state| 
    +------+-----------+-----+ 

：

text_df = sc.parallelize([["Seattle is in Washington.", "a"],["Los Angeles is in California", "b"], 
          ["Banana is a fruit", "c"],["Seattle is not in New Hampshire", "d"], 
          ["California is home to Los Angeles", "e"],["Seattle, California is not a real place.", "f"]])\ 
    .toDF(["body", "source"]) 

代わりに、私たちは、代わりに文字列contains条件と結合を使用します