pySpark mapping multiple columns

by

From my understanding, you can create a map based on columns from reference_df (I assumed this is not a very big dataframe):

map_key = concat_ws('\0', PrimaryLookupAttributeName, PrimaryLookupAttributeValue)
map_value = OutputItemNameByValue

and then use this mapping to get the corresponding values in df1:

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

d = reference_df.agg(collect_set(array(concat_ws('\0','PrimaryLookupAttributeName','PrimaryLookupAttributeValue'), 'OutputItemNameByValue')).alias('m')).first().m
#[['LeaseStatus\x00Abandoned', 'Active'],
# ['LeaseRecoveryType\x00Gross-modified', 'Modified Gross'],
# ['LeaseStatus\x00Archive', 'Expired'],
# ['LeaseStatus\x00Terminated', 'Terminated'],
# ['LeaseRecoveryType\x00Gross w/base year', 'Modified Gross'],
# ['LeaseStatus\x00Draft', 'Pending'],
# ['LeaseRecoveryType\x00Gross', 'Gross']]

mappings = create_map([lit(i) for i in chain.from_iterable(d)])

primaryLookupAttributeName_List = ['LeaseType', 'LeaseRecoveryType', 'LeaseStatus']

df1.select("*", *[ mappings[concat_ws('\0', lit(c), col(c))].alias("Matched[{}]OutputItemNameByValue".format(c)) for c in primaryLookupAttributeName_List ]).show()
+----------------+...+---------------------------------------+-----------------------------------------------+-----------------------------------------+
|SourceSystemName|...|Matched[LeaseType]OutputItemNameByValue|Matched[LeaseRecoveryType]OutputItemNameByValue|Matched[LeaseStatus]OutputItemNameByValue|
+----------------+...+---------------------------------------+-----------------------------------------------+-----------------------------------------+
|          ABC123|...|                                   null|                                          Gross|                               Terminated|
|          ABC123|...|                                   null|                                 Modified Gross|                                  Expired|
|          ABC123|...|                                   null|                                 Modified Gross|                                  Pending|
+----------------+...+---------------------------------------+-----------------------------------------------+-----------------------------------------+

UPDATE: to set Column names from the information retrieved through reference_df dataframe:

# a list of domains to retrieve
primaryLookupAttributeName_List = ['LeaseType', 'LeaseRecoveryType', 'LeaseStatus']

# mapping from domain names to column names: using `reference_df`.`TargetAttributeForName`
NEWprimaryLookupAttributeName_List = dict(reference_df.filter(reference_df['DomainName'].isin(primaryLookupAttributeName_List)).agg(collect_set(array('DomainName', 'TargetAttributeForName')).alias('m')).first().m)

test = dataset_standardFalse2.select("*",*[ mappings[concat_ws('\0', lit(c), col(c))].alias(c_name) for c,c_name in NEWprimaryLookupAttributeName_List.items()])

Note-1: it is better to loop through primaryLookupAttributeName_List so the order of the columns are preserved and in case any entries in primaryLookupAttributeName_List is missing from the dictionary, we can set a default column-name, i.e. Unknown-<col>. In the old method, columns with the missing entries are simply discarded.

test = dataset_standardFalse2.select("*",*[ mappings[concat_ws('\0', lit(c), col(c))].alias(NEWprimaryLookupAttributeName_List.get(c,"Unknown-{}".format(c))) for c in primaryLookupAttributeName_List])

Note-2: per comments, to overwrite the existing column names(untested):

(1) use select:

test = dataset_standardFalse2.select([c for c in dataset_standardFalse2.columns if c not in NEWprimaryLookupAttributeName_List.values()] + [ mappings[concat_ws('\0', lit(c), col(c))].alias(NEWprimaryLookupAttributeName_List.get(c,"Unknown-{}".format(c))) for c in primaryLookupAttributeName_List]).show()

(2) use reduce (not recommended if the List is very long): 

from functools import reduce

df_new = reduce(lambda d, c: d.withColumn(c, mappings[concat_ws('\0', lit(c), col(c))].alias(NEWprimaryLookupAttributeName_List.get(c,"Unknown-{}".format(c)))), primaryLookupAttributeName_List, dataset_standardFalse2)

reference: PySpark create mapping from a dict

More Related Contents:

  1. How to pivot Spark DataFrame?
  2. Filter Spark DataFrame based on another DataFrame that specifies denylist criteria
  3. spark dataframe drop duplicates and keep first
  4. PySpark DataFrame Column Reference: df.col vs. df[‘col’] vs. F.col(‘col’)?
  5. How to make good reproducible Apache Spark examples
  6. Convert list of dictionaries to a pandas DataFrame
  7. How to add a constant column in a Spark DataFrame?
  8. Pandas DataFrame to List of Dictionaries
  9. Spark Dataframe distinguish columns with duplicated name
  10. How do I add a new column to a Spark DataFrame (using PySpark)?
  11. How to perform union on two DataFrames with different amounts of columns in spark?
  12. How to access element of a VectorUDT column in a Spark DataFrame?
  13. Difference between DataFrame, Dataset, and RDD in Spark
  14. Filtering a Pyspark DataFrame with SQL-like IN clause
  15. Updating a dataframe column in spark
  16. Dividing complex rows of dataframe to simple rows in Pyspark
  17. How to create a dictionary of two pandas DataFrame columns
  18. Spark add new column to dataframe with value from previous row
  19. PySpark: multiple conditions in when clause
  20. PySpark create new column with mapping from a dict
  21. Why does Spark think this is a cross / Cartesian join
  22. Pyspark: Replacing value in a column by searching a dictionary
  23. Vectorized “in” function in julia?
  24. Create Spark DataFrame. Can not infer schema for type
  25. Pivot String column on Pyspark Dataframe
  26. _corrupt_record error when reading a JSON file into Spark
  27. dataframe to dict such that one column is the key and the other is the value [duplicate]
  28. Calculating the cosine similarity between all the rows of a dataframe in pyspark
  29. Spark DAG differs with ‘withColumn’ vs ‘select’
  30. How to convert string representation of dictionary in Pandas DataFrame to a new columns?

Leave a Comment