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')]

Как сделать пару ключ: значение из данных внутри столбцов?

например:.

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

Я бы не хотел использовать collect по соображениям производительности. Я пробовал несколько вещей, но не могу получить только значения .

Answer 1

Есть еще один способ преобразовать ваш фрейм данных в 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 2

Как говорит Анкин, вы можете использовать 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"}}']

Answer 3

Вы можете избегать использования udf здесь, используя pyspark.sql.functions.struct и 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 4

да, вы можете использовать

pyspark.sql.types.MapType(keyType, valueType, valueContainsNull=True)

Пожалуйста, поделитесь более подробной информацией, такой как пример выходных данных для фрейма данных и способ, которым вы хотите, в качестве вывода, который поможет в написании фрагмента кода для того же самого.