pyspark根据两列数据创建字典

Question

我有一个带有两列的pyspark数据框：

[Row(zip_code='58542', dma='MIN'),
 Row(zip_code='58701', dma='MIN'),
 Row(zip_code='57632', dma='MIN'),
 Row(zip_code='58734', dma='MIN')]

如何从列内的数据中创建key：value对？

例如：

{
 "58542":"MIN",
 "58701:"MIN",
 etc..
}

由于性能原因，我想避免使用collect。我尝试了一些事情，但似乎不能仅仅得到价值观。

Answer 1

您可以避免在udf此处使用pyspark.sql.functions.structand pyspark.sql.functions.to_json（Spark 2.1及更高版本）：

import pyspark.sql.functions as f
from pyspark.sql import Row

data = [
    Row(zip_code='58542', dma='MIN'),
    Row(zip_code='58701', dma='MIN'),
    Row(zip_code='57632', dma='MIN'),
    Row(zip_code='58734', dma='MIN')
]

df = spark.createDataFrame(data)

df.withColumn("json", f.to_json(f.struct("dma", "zip_code"))).show(truncate=False)
#+---+--------+--------------------------------+
#|dma|zip_code|json                            |
#+---+--------+--------------------------------+
#|MIN|58542   |{"dma":"MIN","zip_code":"58542"}|
#|MIN|58701   |{"dma":"MIN","zip_code":"58701"}|
#|MIN|57632   |{"dma":"MIN","zip_code":"57632"}|
#|MIN|58734   |{"dma":"MIN","zip_code":"58734"}|
#+---+--------+--------------------------------+

如果您希望将zip_code用作键，则可以MapType直接使用创建一个pyspark.sql.functions.create_map：

df.withColumn("json", f.create_map(["zip_code", "dma"])).show(truncate=False)
#+---+--------+-----------------+
#|dma|zip_code|json             |
#+---+--------+-----------------+
#|MIN|58542   |Map(58542 -> MIN)|
#|MIN|58701   |Map(58701 -> MIN)|
#|MIN|57632   |Map(57632 -> MIN)|
#|MIN|58734   |Map(58734 -> MIN)|
#+---+--------+-----------------+

Answer 2

还有另一种方法可以将数据帧转换为 dict。为此，您需要将数据帧转换为键值对 rdd，因为它仅适用于键值对 rdd。因为字典本身是键值对的组合。

data = [
    Row(zip_code='58542', dma='MIN'),
    Row(zip_code='58701', dma='MIN'),
    Row(zip_code='57632', dma='MIN'),
    Row(zip_code='58734', dma='MIN')
]

>>> data.show();
+---+--------+
|dma|zip_code|
+---+--------+
|MIN|   58542|
|MIN|   58701|
|MIN|   57632|
|MIN|   58734|
+---+--------+

将您的数据帧转换为 rdd。

newrdd = data.rdd

由于您希望 zip_code 作为键和 dma 作为值，因此选择 rdd 元素“1”作为键，元素“0”作为值。

keypair_rdd = newrdd.map(lambda x : (x[1],x[0]))

一旦你有了密钥对 rdd 然后只需使用 collectAsMap 将其转换为字典

>>> dict = keypair_rdd.collectAsMap()
>>> print dict
{u'58542': u'MIN', u'57632': u'MIN', u'58734': u'MIN', u'58701': u'MIN'}

>>> dict.keys()
[u'58542', u'57632', u'58734', u'58701']

查找特定键的值：

>>> dict.get('58542')
u'MIN'

Answer 3

与其他解决方案相比，我不确定它的性能如何，但您可以简单地执行以下操作：

dict = {row['zipcode']:row['dma'] for row in df.collect()}
print(dict)
#{'58542': 'MIN', '58701': 'MIN', '57632': 'MIN', '58734': 'MIN'}

Answer 4

正如 Ankin 所说，您可以使用 MapType 来实现此目的：

import pyspark
from pyspark.sql import Row

sc = pyspark.SparkContext()
spark = pyspark.sql.SparkSession(sc)

data = spark.createDataFrame([Row(zip_code='58542', dma='MIN'),
 Row(zip_code='58701', dma='MIN'),
 Row(zip_code='57632', dma='MIN'),
 Row(zip_code='58734', dma='MIN')])

data.show()

输出：

+---+--------+
|dma|zip_code|
+---+--------+
|MIN|   58542|
|MIN|   58701|
|MIN|   57632|
|MIN|   58734|
+---+--------+


from pyspark.sql.functions import udf
from pyspark.sql import types as T

@udf(T.MapType(T.StringType(), T.StringType()))
def create_struct(zip_code, dma):
    return {zip_code: dma}

data.withColumn('struct', create_struct(data.zip_code, data.dma)).toJSON().collect()

输出：

['{"dma":"MIN","zip_code":"58542","struct":{"58542":"MIN"}}',
 '{"dma":"MIN","zip_code":"58701","struct":{"58701":"MIN"}}',
 '{"dma":"MIN","zip_code":"57632","struct":{"57632":"MIN"}}',
 '{"dma":"MIN","zip_code":"58734","struct":{"58734":"MIN"}}']