Конвертировать фрейм данных Pyspark в словарь

Question

Я пытаюсь преобразовать фрейм данных Pyspark в словарь.

Вот пример файла CSV -

Col0, Col1
-----------
A153534,BDBM40705
R440060,BDBM31728
P440245,BDBM50445050

Я пришел с этим кодом -

from rdkit import Chem
from pyspark import SparkContext
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession

sc = SparkContext.getOrCreate()
spark = SparkSession(sc)

df = spark.read.csv("gs://my-bucket/my_file.csv") # has two columns

# Creating list
to_list = map(lambda row: row.asDict(), df.collect())

#Creating dictionary
to_dict = {x['col0']: x for x in to_list }

Это создает словарь, как показано ниже -

'A153534': {'col0': 'A153534', 'col1': 'BDBM40705'}, 'R440060': {'col0': 'R440060', 'col1': 'BDBM31728'}, 'P440245': {'col0': 'P440245', 'col1': 'BDBM50445050'} 

Но я хочу вот такой словарь -

{'A153534': 'BDBM40705'}, {'R440060': 'BDBM31728'}, {'P440245': 'BDBM50445050'}

Как мне это сделать?

Я попробовал rdd Решение от Йоло, но я получаю ошибку. Подскажите, пожалуйста, что я делаю неправильно?

py4j.protocol.Py4JError: Произошла ошибка при вызове o80.isBarrier. Трассировка: py4j.Py4JException: Метод isBarrier ([]) не существует в py4j.reflection.ReflectionEngine.getMethod (ReflectionEngine. java: 318) в py4j.reflection.ReflectionEngine.getMethod (ReflectionEngine. java 326): py4j.Gateway.invoke (Gateway. java: 274) на py4j.commands.AbstractCommand.invokeMethod (AbstractCommand. java: 132) на py4j.commands.CallCommand.execute (CallCommand. java: 79) на py4j. .GatewayConnection.run (GatewayConnection. java: 238) в java .lang.Thread.run (Thread. java: 748)

Answer 1

Вот способ сделать это, используя rdd:

df.rdd.map(lambda x: {x.Col0: x.Col1}).collect()

[{'A153534': 'BDBM40705'}, {'R440060': 'BDBM31728'}, {'P440245': 'BDBM50445050'}]

Answer 2

Это может помочь вам:

df = spark.read.csv('/FileStore/tables/Create_dict.txt',header=True)

df = df.withColumn('dict',to_json(create_map(df.Col0,df.Col1)))

df_list = [row['dict'] for row in df.select('dict').collect()]

df_list

Вывод:

['{"A153534":"BDBM40705"}',
 '{"R440060":"BDBM31728"}',
 '{"P440245":"BDBM50445050"}']