我有兩個 Spark 資料框，如下所示：
> cities_df
---------- ---------------------------
| city_id| cities|
---------- ---------------------------
| 22 |[Milan, Turin, Rome] |
---------- ---------------------------
| 15 |[Naples, Florence, Genoa] |
---------- ---------------------------
| 43 |[Houston, San Jose, Boston]|
---------- ---------------------------
| 56 |[New York, Dallas, Chicago]|
---------- ---------------------------
> countries_df
---------- ----------------------------------
|country_id| countries|
---------- ----------------------------------
| 680 |{'country': [56, 43], 'add': []} |
---------- ----------------------------------
| 11 |{'country': [22, 15], 'add': [32]}|
---------- ----------------------------------
中的國家/地區值countries_df是資料框中的城市 ID cities_df。
我需要合并這些資料框以用資料country框中的值替換城市 ID cities_df。
預期輸出：
| country_id | 國家 | 分組城市 |
|---|---|---|
| 680 | {'國家'：[56, 43]，'添加'：[]} | [紐約、達拉斯、芝加哥、休斯頓、圣何塞、波士頓] |
| 11 | {'國家'：[22, 15]，'添加'：[32]} | [米蘭、都靈、羅馬、那不勒斯、佛羅倫薩、熱那亞] |
獲取grouped_cities的值不一定是陣列型別，可以是字串。
如何使用 PySpark 獲得此結果？
uj5u.com熱心網友回復：
輸入：
from pyspark.sql import functions as F
cities_df = spark.createDataFrame(
[(22, ['Milan', 'Turin', 'Rome']),
(15, ['Naples', 'Florence', 'Genoa']),
(43, ['Houston', 'San Jose', 'Boston']),
(56, ['New York', 'Dallas', 'Chicago'])],
['city_id', 'cities']
)
countries_df = spark.createDataFrame(
[(680, {'country': [56, 43], 'add': []}),
(11, {'country': [22, 15], 'add': [32]})],
['country_id', 'countries']
)
腳本：
df_expl = countries_df.withColumn('city_id', F.explode('countries.country'))
df_joined = df_expl.join(cities_df, 'city_id', 'left')
df = df_joined.groupBy('country_id').agg(
F.first('countries').alias('countries'),
F.flatten(F.collect_list('cities')).alias('grouped_cities')
)
df.show(truncate=0)
# ---------- ---------------------------------- ------------------------------------------------------
# |country_id|countries |grouped_cities |
# ---------- ---------------------------------- ------------------------------------------------------
# |11 |{add -> [32], country -> [22, 15]}|[Naples, Florence, Genoa, Milan, Turin, Rome] |
# |680 |{add -> [], country -> [56, 43]} |[Houston, San Jose, Boston, New York, Dallas, Chicago]|
# ---------- ---------------------------------- ------------------------------------------------------
uj5u.com熱心網友回復：
另一種方法。使用 select 在 countries_df 上創建一個新列。Groupby 使用 country_id，并將國家列轉換為字串。代碼如下。
new =cities_df.join(countries_df.select('*',explode('countries.country').alias('city_id')), how='left', on='city_id').groupby('country_id',col('countries').cast('string').alias('countries')).agg(flatten(collect_set('cities')).alias('cities')).show(truncate=False)
---------- ---------------------------------- ------------------------------------------------------
|country_id|countries |cities |
---------- ---------------------------------- ------------------------------------------------------
|11 |{add -> [32], country -> [22, 15]}|[Milan, Turin, Rome, Naples, Florence, Genoa] |
|680 |{add -> [], country -> [56, 43]} |[New York, Dallas, Chicago, Houston, San Jose, Boston]|
---------- ---------------------------------- ------------------------------------------------------
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/481492.html
標籤：数组 阿帕奇火花 pyspark 合并 apache-spark-sql