Spark数组列中的值如何高效地替换为来自Pandas数据框的值？

huangapple
go评论103阅读模式
英文:

How can values in a Spark array column be efficiently replaced with values from a Pandas data frame?

问题

I understand your request. Here's the translated code portion:

  1. 我有一个包含销售篮子中产品ID数组的Spark数据框。
  3. df_baskets = spark.createDataFrame(
  4.     [(1, ["546", "689", "946"]), (2, ["546", "799"] )],
  5.     ("case_id","basket")
  6. )
  8. 产品数据 = pd.DataFrame({
  9.   "product_id": ["546", "689", "946", "799"],
  10.   "new_product_id": ["S12", "S74", "S34", "S56"]
  11.   })
  13. def 获取新ID(product_id: str) -> str:
  14.   try:
  15.     行 = 产品数据[产品数据["product_id"] == product_id]
  16.     return 行["new_product_id"].item()
  17.   except ValueError:
  18.     return product_id
  20. 应用获取新ID = F.udf(lambda basket: [获取新ID(product) for product in basket], T.ArrayType(T.StringType()))
  22. df_baskets = (
  23.   df_baskets
  24.     .withColumn('basket_renamed', 应用获取新ID(F.col('basket')))
  25. )
  27. df_baskets.show()
  29. #+-------+---------------+---------------+
  30. #|case_id|         basket| basket_renamed|
  31. #+-------+---------------+---------------+
  32. #|      1|["546", "689", "946"]|["S12", "S74", "S34"]|
  33. #|      2|     ["546", "799"]|     ["S12", "S56"]|
  34. #+-------+---------------+---------------+

Please note that I have translated the code as per your instructions. Let me know if you need any further assistance!

英文:

I have a Spark data frame that contains a column of arrays with product ids from sold baskets.

  1. import pandas as pd 
  2. import pyspark.sql.types as T
  3. from pyspark.sql import functions as F
  5. df_baskets = spark.createDataFrame(
  6.     [(1, ["546", "689", "946"]), (2, ["546", "799"] )],
  7.     ("case_id","basket")
  8. )
  10. df_baskets.show()
  12. #+-------+---------------+
  13. #|case_id|         basket|
  14. #+-------+---------------+
  15. #|      1|[546, 689, 946]|
  16. #|      2|     [546, 799]|
  17. #+-------+---------------+

I would like to replace the product ids in each array with new ids given in a pandas data frame.

  2. product_data = pd.DataFrame({
  3.   "product_id": ["546", "689", "946", "799"],
  4.   "new_product_id": ["S12", "S74", "S34", "S56"]
  5.   })
  7. product_data

I was able to replace the values by applying a simple python function to the column that performs a lookup on the python data frame.

  2. def get_new_id(product_id: str) -> str:
  3.   try:
  4.     row = product_data[product_data["product_id"] == product_id]
  5.     return row["new_product_id"].item()
  6.   except ValueError:
  7.     return product_id
  9. apply_get = F.udf(lambda basket: [get_new_id(product) for product in basket], T.ArrayType(T.StringType()))
  11. df_baskets = (
  12.   df_baskets
  13.     .withColumn('basket_renamed', apply_get(F.col('basket')))
  14. )
  16. df_baskets.show()
  18. #+-------+---------------+---------------+
  19. #|case_id|         basket| basket_renamed|
  20. #+-------+---------------+---------------+
  21. #|      1|[546, 689, 946]|[S12, S74, S34]|
  22. #|      2|     [546, 799]|     [S12, S56]|
  23. #+-------+---------------+---------------+

However, this approach has proven to be quite slow in data frames containing several tens of millions of cases. Is there more efficient way to do this replacement (e.g. by using a different data structure than a pandas data frame or a different method)?

答案1

得分: 2

以下是您要翻译的内容：

您可以将原始数据拆分并加入到product_data（在将其转换为Spark数据框之后）：

  1. (
  2.     df_baskets
  3.     .withColumn("basket", F.explode(F.col("basket")))
  4.     .join(
  5.         spark.createDataFrame(product_data)
  6.         .withColumnRenamed("product_id", "basket")
  7.         .withColumnRenamed("new_product_id", "basket_renamed"),
  8.         on="basket"
  9.     )
  10.     .groupby("case_id")
  11.     .agg(
  12.         F.collect_list(F.col("basket")).alias("basket"),
  13.         F.collect_list(F.col("basket_renamed")).alias("basket_renamed")
  14.     )
  15. ).show()

输出：

  1. |case_id|         basket| basket_renamed|
  2. +-------+---------------+---------------+
  3. |      1|[546, 689, 946]|[S12, S74, S34]|
  4. |      2|     [546, 799]|     [S12, S56]|
  5. +-------+---------------+---------------+
英文:

You could explode your original data and join on product_data (after converting it to a spark frame)

  1. (
  2.     df_baskets
  3.     .withColumn("basket", F.explode(F.col("basket")))
  4.     .join(
  5.         spark.createDataFrame(product_data)
  6.         .withColumnRenamed("product_id", "basket")
  7.         .withColumnRenamed("new_product_id", "basket_renamed"),
  8.         on="basket"
  9.     )
  10.     .groupby("case_id")
  11.     .agg(
  12.         F.collect_list(F.col("basket")).alias("basket"),
  13.         F.collect_list(F.col("basket_renamed")).alias("basket_renamed")
  14.     )
  15. ).show()

Output:

  1. |case_id|         basket| basket_renamed|
  2. +-------+---------------+---------------+
  3. |      1|[546, 689, 946]|[S12, S74, S34]|
  4. |      2|     [546, 799]|     [S12, S56]|
  5. +-------+---------------+---------------+

答案2

得分: 2

你可以使用RDD和map。

将pandas dataframe的行转换为{old: new}值的字典，然后在RDD中使用map来获取映射的new_product_id。

这是一个示例：

  1. # 将pandas df转换为字典
  2. # 还可以以其他方式完成
  3. old_new_id_dict = {}
  5. for i in range(len(product_data_df)):
  6.     old_new_id_dict[product_data_df.loc[i, 'product_id']] = product_data_df.loc[i, 'new_product_id']
  8. # {'546': 'S12', '689': 'S74', '946': 'S34', '799': 'S56'}
  10. old_new_id_dict_bc = spark.sparkContext.broadcast(old_new_id_dict)
  12. # 使用`map`处理值
  13. data_sdf.rdd. \
  14.     map(lambda r: (r.case_id, r.basket, [old_new_id_dict_bc.value[k] for k in r.basket])). \
  15.     toDF(['case_id', 'basket', 'new_basket']). \
  16.     show()
  18. # +-------+---------------+---------------+
  19. # |case_id|         basket|     new_basket|
  20. # +-------+---------------+---------------+
  21. # |      1|[546, 689, 946]|[S12, S74, S34]|
  22. # |      2|     [546, 799]|     [S12, S56]|
  23. # +-------+---------------+---------------+

希望这能帮助你。

英文:

you could use RDD and map.

convert the pandas dataframe rows to a dict of {old: new} values. then use a map in RDD to fetch the mapped new_product_id.

here's an example

  1. # convert pandas df to dict
  2. # can be done in other ways as well
  3. old_new_id_dict = {}
  5. for i in range(len(product_data_df)):
  6.     old_new_id_dict[product_data_df.loc[i, 'product_id']] = product_data_df.loc[i, 'new_product_id']
  8. # {'546': 'S12', '689': 'S74', '946': 'S34', '799': 'S56'}
  10. old_new_id_dict_bc = spark.sparkContext.broadcast(old_new_id_dict)
  12. # `map` the values
  13. data_sdf.rdd. \
  14.     map(lambda r: (r.case_id, r.basket, [old_new_id_dict_bc.value[k] for k in r.basket])). \
  15.     toDF(['case_id', 'basket', 'new_basket']). \
  16.     show()
  18. # +-------+---------------+---------------+
  19. # |case_id|         basket|     new_basket|
  20. # +-------+---------------+---------------+
  21. # |      1|[546, 689, 946]|[S12, S74, S34]|
  22. # |      2|     [546, 799]|     [S12, S56]|
  23. # +-------+---------------+---------------+

huangapple
  • 本文由 发表于 2023年4月13日 20:10:14
  • 转载请务必保留本文链接：https://go.coder-hub.com/76005256.html
  • apache-spark
  • pandas
  • pyspark
  • python
  • scala
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定