PySpark 模式匹配和分配关联值

Question

df1：

campaign_name   campaign_team
einsurancep09   other
estoreemicardcdwpnov06  other
estoreemicardwmnov06    other
estoreemicardgenericspnov06 other

df2：

terms   product_category    product
insurance   insurance   null
def emi store
ab  bhi asd
de  lic cards
a   credit  cards

以下是我的场景：

1. 'terms'列（df2）按字符串长度降序排列。
2. 应与contains/like 的df1 的campaign_name 进行比较。
3. 无论terms 字符串matches first 与campaign_name，其product_category 和product 都应被拾取并应作为新列添加到df1 中。
4. 对于campaign_name 值“einsurancep09”，来自terms 的“insurance”值包含在campaign_name 中，因此它的product_category 和product 被拾取并作为df1 添加到输出中。
5. 另一个例子：考虑其余3条记录，其中containdef、ab和de在campaign_name字符串中，但我们选择product_category和“def”的产品作为appeared first与“ab”和“de”相比，是longest in the length

下面是我的代码：

df1 = df1.withColumn("product_category",when(df1.campaign_name.contains(df2.terms),df2.product_category).otherwise('other'))

但是，它给了我以下错误：

   raise converted from None
pyspark.sql.utils.AnalysisException: Resolved attribute(s) terms#37,product_category#38 missing from campaign_name#16,campaign_team#17 in operator !Project [campaign_name#16, campaign_team#17, CASE WHEN Contains(campaign_name#16, terms#37) THEN product_category#38 ELSE other END AS product_category#44].;
!Project [campaign_name#16, campaign_team#17, CASE WHEN Contains(campaign_name#16, terms#37) THEN product_category#38 ELSE other END AS product_category#44]
+- Relation[campaign_name#16,campaign_team#17] csv

那么我哪里错了？

根据堆栈的回答，我得到以下输出：

+---------------+-------------+---------+----------------+-------+
|campaign_name  |campaign_team|terms    |product_category|product|
+---------------+-------------+---------+----------------+-------+
|einsurancepnm06|other        |insurance|Insurance       |NaN    |
+---------------+-------------+---------+----------------+-------+

预期输出：

Answer 1

如果数据帧df2 的容量很大，那么 spark 将不会被优化以执行此类操作。如果不是（

• 交叉加入
• UDF

方法一：交叉连接

步骤：

1. 添加新列rw 并按照df2 中的要求顺序分配行号。

2. 交叉连接df1 并排序df2 并创建一个新的数据框df3。

3. 使用包含列函数并添加新列match。

4. 选择分区后的第一行按所有df1列并按df2行号和match列排序

代码

>>> df1.show(truncate=False)
+---------------------------+-------------+
|campaign_name              |campaign_team|
+---------------------------+-------------+
|einsurancep09              |other        |
|estoreemicardcdwpnov06     |other        |
|estoreemicardwmnov06       |other        |
|estoreemicardgenericspnov06|other        |
|abcdefcdwpnov06            |other        |
|abcdefwmnov06              |other        |
|abcdefgenericspnov06       |other        |
+---------------------------+-------------+

>>> df2.show(truncate=False)
+---------+----------------+-------+---+
|terms    |product_category|product|rw |
+---------+----------------+-------+---+
|insurance|insurance       |null   |1  |
|def      |emi             |store  |2  |
|ab       |bhi             |asd    |3  |
|de       |lic             |cards  |4  |
|a        |credit          |cards  |5  |
+---------+----------------+-------+---+

>>> df3 = df1.crossJoin(df2)
>>> df4 = df3.withColumn("match", col("campaign_name").contains(col("terms")))
>>> W = Window.partitionBy(col("campaign_name"), col("campaign_team")).orderBy(col("match").desc(), col("rw"))

>>> finalDF = df4.withColumn("rn", row_number().over(W)).filter(col("rn") == lit(1)).drop("rn", "terms","rw","match")
>>> finalDF.show(truncate=False)
+---------------------------+-------------+----------------+-------+
|campaign_name              |campaign_team|product_category|product|
+---------------------------+-------------+----------------+-------+
|estoreemicardgenericspnov06|other        |credit          |cards  |
|abcdefwmnov06              |other        |emi             |store  |
|estoreemicardcdwpnov06     |other        |credit          |cards  |
|estoreemicardwmnov06       |other        |credit          |cards  |
|einsurancep09              |other        |insurance       |null   |
|abcdefgenericspnov06       |other        |emi             |store  |
|abcdefcdwpnov06            |other        |emi             |store  |
+---------------------------+-------------+----------------+-------+

方法 2 UDF

步骤：

1. 添加新列rw并按照df2中的要求顺序分配行号。
2. 根据行号将所有df2 行收集到一个行中并创建一个新的Dataframe df3。
3. 使用下面的任何方法并创建一个新的 Dataframe df4。
   • 将df3 行转换为字符串变量，并将字符串文字添加为df1 中的新列。 或者
   • 交叉加入df1 和df3。
4. 如下声明UDF。
5. 调用UDF并添加一个带有返回值的新列。
6. 根据输出返回创建要求列。

代码

>>> df3 = df2.na.fill("").groupBy(lit(1)).agg(sort_array(collect_list(concat(col("rw"),lit(":"), col("terms"), lit(":"), col("product_category"), lit(":"), col("product")))).alias("Check")).withColumn("Check", concat_ws(",", col("Check"))).drop("1")
>>> df3.show(truncate=False)
+-----------------------------------------------------------------------------------+
|Check                                                                              |
+-----------------------------------------------------------------------------------+
|1:insurance:insurance:,2:def:emi:store,3:ab:bhi:asd,4:de:lic:cards,5:a:credit:cards|
+-----------------------------------------------------------------------------------+

>>> df4 = df1.crossJoin(df3)
>>> def categoryFunction(name, Check):
        checkList = Check.lower().split(",")
        out = ""
        match = False
        for Key in checkList:
            keyword = Key.split(":",2)
            terms = keyword[1]
            tempOut = keyword[2]
            if terms in name.lower():
                out = tempOut
                match = True
            if match:
                break
        return out
   
>>> categoryUDF = udf(categoryFunction, StringType())
>>> finalDF = df4.withColumn("out", categoryUDF(col("campaign_name"), col("Check"))).drop("Check").withColumn("out", split(col("out"), ":")).withColumn("product_category", col("out")[0]).withColumn("product", col("out")[1]).drop("out").show(truncate=False)
>>> finalDF.show(truncate=False)
+---------------------------+-------------+----------------+-------+
|campaign_name              |campaign_team|product_category|product|
+---------------------------+-------------+----------------+-------+
|einsurancep09              |other        |insurance       |       |
|estoreemicardcdwpnov06     |other        |credit          |cards  |
|estoreemicardwmnov06       |other        |credit          |cards  |
|estoreemicardgenericspnov06|other        |credit          |cards  |
|abcdefcdwpnov06            |other        |emi             |store  |
|abcdefwmnov06              |other        |emi             |store  |
|abcdefgenericspnov06       |other        |emi             |store  |
+---------------------------+-------------+----------------+-------+

Answer 2

假设

数据集 df1 应具有满足 OP 要求的顺序。所以我介绍rec_no列

df = spark.sql("""
select 'abcdefcdwpnovo6' campaign_name, 'other' campaign_team union all
select 'abcdefdwpnovo6' , 'other' union all
select 'abcdefgenericpnovo6' , 'other' 
""")
df.createOrReplaceTempView("df")
df.show()

+-------------------+-------------+
|      campaign_name|campaign_team|
+-------------------+-------------+
|    abcdefcdwpnovo6|        other|
|     abcdefdwpnovo6|        other|
|abcdefgenericpnovo6|        other|
+-------------------+-------------+

df1 = spark.sql("""
select 1 rec_no, 'def' terms, 'emi' product_category, 'store' product union all 
select 2, 'abc' ,'bhi' ,'asd' union all
select 3, 'de' ,'lic' ,'cards' union all
select 4, 'a' ,'credit' ,'cards' 
""")
df1.createOrReplaceTempView("df1")
df1.show()

+------+-----+----------------+-------+
|rec_no|terms|product_category|product|
+------+-----+----------------+-------+
|     1|  def|             emi|  store|
|     2|  abc|             bhi|    asd|
|     3|   de|             lic|  cards|
|     4|    a|          credit|  cards|
+------+-----+----------------+-------+

输出：

您可以删除 ps、rk 和 rec_no 列

spark.sql("""
with t1 ( select * from df a cross join df1 b ),
     t2 ( select rec_no, campaign_name,campaign_team,terms,product_category,product x, position(terms,campaign_name) ps,
    rank() over(order by rec_no) rk from t1  where position(terms,campaign_name)>0 )
    select * from t2 where rk=1
""").show()

+------+-------------------+-------------+-----+----------------+-----+---+---+
|rec_no|      campaign_name|campaign_team|terms|product_category|    x| ps| rk|
+------+-------------------+-------------+-----+----------------+-----+---+---+
|     1|    abcdefcdwpnovo6|        other|  def|             emi|store|  4|  1|
|     1|     abcdefdwpnovo6|        other|  def|             emi|store|  4|  1|
|     1|abcdefgenericpnovo6|        other|  def|             emi|store|  4|  1|
+------+-------------------+-------------+-----+----------------+-----+---+---+

更新 1：

OP的问题仍然不清楚。试试下面。

spark.sql("""
with t1 ( select * from df a cross join df1 b ),
     t2 ( select rec_no, campaign_name,campaign_team,terms,product_category,product, position(product_category,campaign_name) ps,
    rank() over(partition by product_category order by rec_no) rk from t1 where position(product_category,campaign_name)>0 
    )
    select * from t2 where rk=1 order by rec_no
""").show(truncate=False)

+------+---------------------------+-------------+---------+----------------+-------+---+---+
|rec_no|campaign_name              |campaign_team|terms    |product_category|product|ps |rk |
+------+---------------------------+-------------+---------+----------------+-------+---+---+
|1     |einsurancep09              |other        |insurance|insurance       |null   |2  |1  |
|2     |estoreemicardcdwpnov06     |other        |def      |emi             |store  |7  |1  |
|2     |estoreemicardgenericspnov06|other        |def      |emi             |store  |7  |1  |
|2     |estoreemicardwmnov06       |other        |def      |emi             |store  |7  |1  |
+------+---------------------------+-------------+---------+----------------+-------+---+---+