将字符串列转换为pyspark sql数据框中的字典答案

【问题标题】：Convert column of strings to dictionaries in pyspark sql dataframe将字符串列转换为pyspark sql数据框中的字典
【发布时间】：2020-03-02 12:05:26
【问题描述】：

我必须使用每行都是一个 json 对象的文件格式。例如：

{'Attribute 1': 'A', 'Attribute 2': 1.5, 'Attribute 3': ['A','B','C'], 'Attribute 4': {'A': 5}}
{'Attribute 1': 'B', 'Attribute 2': 2.0, 'Attribute 3': ['A'], 'Attribute 4': {'A': 4}}
{'Attribute 1': 'C', 'Attribute 2': 1.7, 'Attribute 3': ['A','C'], 'Attribute 4': {'A': 3}}

请注意，这不是有效的 json 文件格式，因为它没有包含在数组中。而且，实际的结构要大得多，嵌套的也多。这些文件分布在 s3 中。我之前只使用过 parquet 或 csv，所以我不确定如何读取这些文件。

我目前正在编写一个流程来将此数据与其他几个表连接起来，并且由于数据很大并且位于 s3 中，我正在使用 emr 集群中的 pyspark.sql 来执行操作。我可以使用以下方法创建一个包含对象作为字符串的单列的表：

from pyspark.sql import SQLContext
from pyspark.sql.types import StructType, StructField, StringType
sqlContext = SQLContext(sc)

schema = StructType([
    StructField('json_format', StringType())
])

context = sqlContext.read
context = context.schema(schema)

df = context.load(
    folder_path,
    format='com.databricks.spark.csv',
    delimiter=','
)
df.createOrReplaceTempView('my_table')

如何将此列转换为可以访问各种属性的字典？是否有等效的 lambda 函数？

【问题讨论】：

标签： json string pyspark pyspark-sql

【解决方案1】：

要生成有效的 json 对象，我们可以将所有 ' 替换为 "，然后使用 get_json_object() 函数我们可以访问属性。

Example:

df=sqlContext.sql("""select string("{'Attribute1': 'A', 'Attribute 2': 1.5, 'Attribute 3': ['A','B','C'], 'Attribute 4': {'A': 5}}") as str""")

#replacing ' with " using regexp_replace
df=df.withColumn("str",regexp_replace(col("str"),"\'","\""))
df.show(10,False)

#+----------------------------------------------------------------------------------------------+
#|str                                                                                           |
#+----------------------------------------------------------------------------------------------+
#|{"Attribute1": "A", "Attribute 2": 1.5, "Attribute 3": ["A","B","C"], "Attribute 4": {"A": 5}}|
#+----------------------------------------------------------------------------------------------+

#registering temp table
df.registerTempTable("tt")

#accessing Attribute 3
sqlContext.sql("select get_json_object(str,'$.Attribute 3') from tt").show()
#+-------------+
#|          _c0|
#+-------------+
#|["A","B","C"]|
#+-------------+

#accessing first array element from Attribute 3
sqlContext.sql("select get_json_object(str,'$.Attribute 3[0]') from tt").show()
#+---+
#|_c0|
#+---+
#|  A|
#+---+

#accessing Attribute 2
sqlContext.sql("select get_json_object(str,'$.Attribute 2') from tt").show()
#+---+
#|_c0|
#+---+
#|1.5|
#+---+

【讨论】：