pySpark 映射多个变量

Question

下面的代码将我的参考 df 的值和列名映射到我的实际数据集，查找完全匹配，如果找到完全匹配，则返回 OutputValue。但是，我正在尝试添加规则，即当PrimaryValue = DEFAULT 也返回OutputValue。

我正在尝试解决此问题的解决方案是创建一个具有空值的新数据框 - 因为下面的代码没有提供匹配项。因此，下一步将针对空值，其对应的PrimaryValue = DEFAULT 将空值替换为OutputValue。

  #create a map based on columns from reference_df
  map_key = concat_ws('\0', final_reference.PrimaryName, final_reference.PrimaryValue)
  map_value = final_reference.OutputValue

  #dataframe of concatinated mappings to get the corresponding OutputValues from reference table
  d = final_reference.agg(collect_set(array(concat_ws('\0','PrimaryName','PrimaryValue'), 'OutputValue')).alias('m')).first().m
  #display(d)

  #iterate through mapped values 
  mappings = create_map([lit(i) for i in chain.from_iterable(d)])

  #dataframe with corresponding matched OutputValues
  dataset = datasetM.select("*",*[ mappings[concat_ws('\0', lit(c), col(c))].alias(c_name) for c,c_name in matched_List.items()]) 
  display(dataset)

Answer 1

根据 cmets 中的讨论，我认为您只需从现有映射中添加一个默认映射，然后使用 coalease() 函数查找第一个非空值，如下所示：

from pyspark.sql.functions import collect_set, array, concat_ws, lit, col, create_map, coalesce

# skip some old code

d    
#[['LeaseStatus\x00Abandoned', 'Active'],
# ['LeaseStatus\x00DEFAULT', 'Pending'],
# ['LeaseRecoveryType\x00Gross-modified', 'Modified Gross'],
# ['LeaseStatus\x00Archive', 'Expired'],
# ['LeaseStatus\x00Terminated', 'Terminated'],
# ['LeaseRecoveryType\x00Gross w/base year', 'Modified Gross'],
# ['LeaseRecoveryType\x00Gross', 'Gross']]

# original mapping
mappings = create_map([ lit(j) for i in d for j in i ])

# default mapping
mappings_default = create_map([ lit(j.split('\0')[0]) for i in d if i[0].upper().endswith('\x00DEFAULT') for j in i ])
#Column<b'map(LeaseStatus, Pending)'>

# a set of available PrimaryLookupAttributeName
available_list = set([ i[0].split('\0')[0] for i in d ])
# {'LeaseRecoveryType', 'LeaseStatus'}

# use coalesce to find the first non-null values from mappings, mappings_defaul etc
datasetPrimaryAttributes_False = datasetMatchedPortfolio.select("*",*[ 
  coalesce(
    mappings[concat_ws('\0', lit(c), col(c))],
    mappings_default[c],
    lit("Not Specified at Source" if c in available_list else "Lookup not found")
  ).alias(c_name) for c,c_name in matchedAttributeName_List.items()])

一些解释：

(1) d 是从reference_df 中检索到的列表列表，我们使用列表推导[ lit(j) for i in d for j in i ] 将其展平为列表并将展平后的列表应用于create_map 函数：

(2) mappings_default 与上面类似，但添加了一个if 条件作为过滤器，并且只保留PrimaryLookupAttributeValue（这是内部列表i[0] 的第一项）以@987654327 结尾的条目@ 然后使用 split 从 map_key 中剥离 PrimaryLookupAttributeValue（基本上是 \x00DEFAULT）。