PySpark - переименовывать имена ключей в JSON, хранящиеся в виде строкового столбца в файле CSV

Question

Я хочу обновить имя ключа в моем json, сохраненном в виде строкового столбца, и сохранить его обратно в виде столбца строкового типа. Я читаю такие столбцы из моего csv и сохраняю их как csv.

Вот так выглядит мой ввод csv.

candidate_email,transactions
cust2@email.com,"[{'transaction_id':'12', 'transaction_amount':'$23.43'},{'transaction_id':'15', 'transaction_amount':'$723.41'}]"
cust1@email.com,"[{'transaction_id':'10', 'transaction_amount':'$55.99'},{'transaction_id':'11', 'transaction_amount':'$20.46'},{'transaction_id':'13', 'transaction_amount':'$5.89'},{'transaction_id':'14', 'transaction_amount':'$35.61'}]"

Я хочу заменить transaction_id ключ на id и transaction_amount на amount в моем json и сохранить его как csv.

input_df = spark.read.csv('transactions/*.csv', header='true', inferSchema = True)
input_df.printSchema()
# root
#  |-- candidate_email: string (nullable = true)
#  |-- transactions: string (nullable = true)

input_df.show(10, False)
# +-----------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
# |candidate_email|transactions                                                                                                                                                                                                                |
# +-----------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
# |cust2@email.com        |[{'transaction_id':'12', 'transaction_amount':'$23.43'},{'transaction_id':'15', 'transaction_amount':'$723.41'}]                                                                                                            |
# |cust1@email.com        |[{'transaction_id':'10', 'transaction_amount':'$55.99'},{'transaction_id':'11', 'transaction_amount':'$20.46'},{'transaction_id':'13', 'transaction_amount':'$5.89'},{'transaction_id':'14', 'transaction_amount':'$35.61'}]|
# +-----------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

Как заменить ключи, чтобы получить следующий вывод?

output_df.show(10,False)
# +---------------+----------------------------------------------------------------------------------------------------------------------------+
# |candidate_email|transactions                                                                                                                |
# +---------------+----------------------------------------------------------------------------------------------------------------------------+
# |cust1@email.com|[{'id':'10', 'amount':'$55.99'},{'id':'11', 'amount':'$20.46'},{'id':'13', 'amount':'$5.89'},{'id':'14', 'amount':'$35.61'}]|
# |cust2@email.com|[{'id':'12', 'amount':'$23.43'},{'id':'15', 'amount':'$723.41'}]                                                            |
# +---------------+----------------------------------------------------------------------------------------------------------------------------+

Примечание: оба столбца являются столбцами строкового типа.

output_df.printSchema()
# root
#  |-- candidate_email: string (nullable = true)
#  |-- transactions: string (nullable = true)

Answer 1

Используйте from_json, чтобы прочитать столбец транзакций как array(struct...), а затем привести к нужным именам полей.

• затем explode + to_json + groupBy + collect_list, чтобы получить требуется json.

Example:

df.show()
#+---------------+----------------------------------------------------------------------------------------------------------------+
#|candidate_email|transactions                                                                                                    |
#+---------------+----------------------------------------------------------------------------------------------------------------+
#|cust2@email.com|[{'transaction_id':'12', 'transaction_amount':'$23.43'},{'transaction_id':'15', 'transaction_amount':'$723.41'}]|
#+---------------+----------------------------------------------------------------------------------------------------------------+

st=ArrayType(StructType([StructField("transaction_id", StringType()),StructField("transaction_amount", StringType())]))

df.withColumn("jsn",from_json(col("transactions"),st).cast("array<struct<id:string,amount:string>>")).\
selectExpr("*","explode(jsn)").\
select("*","col.*").\
drop(*drop_cols).\
selectExpr("candidate_email","to_json(struct(id,amount)) as trans").\
groupBy("candidate_email").\
agg(collect_list("trans").alias("transactions")).\
show(10,False)

#+---------------+---------------------------------------------------------------+
#|candidate_email|transactions                                                   |
#+---------------+---------------------------------------------------------------+
#|cust2@email.com|[{"id":"12","amount":"$23.43"}, {"id":"15","amount":"$723.41"}]|
#+---------------+---------------------------------------------------------------+