从 spark table 中提取值 json 给出 SyntaxError 错误或 keyType should be DataType 错误

Question

假设我下面有这样的数据框org：

id |raw
123|{"inn":"123”, "prof": "tkie"}

我需要从 raw 列的 json 中提取所有 ids 值到新列中，该怎么做？

我试过了：

org.withColumn('inn', from_json($"raw", MapType(StringType, StringType))).withColumn('inn', col('searchcard'.getItem('inn')))

出现错误：

File "", line 1 org.withColumn('inn', from_json($"raw", MapType(StringType, StringType))).withColumn('inn', col('searchcard'.getItem('inn'))) ^ SyntaxError: invalid syntax

并且：

org.withColumn('inn', from_json("raw", MapType(StringType, StringType))).withColumn('inn', col('searchcard'.getItem('inn')))

错误：

AssertionError Traceback (most recent call last) in () ----> 1 org.withColumn('inn', from_json("raw", MapType(StringType, StringType))).withColumn('inn', col('searchcard'.getItem('inn')))
/opt/cloudera/parcels/SPARK2/lib/spark2/python/pyspark/sql/types.py in init(self, keyType, valueType, valueContainsNull) 342 False 343 """ --> 344 assert isinstance(keyType, DataType), "keyType should be DataType" 345 assert isinstance(valueType, DataType), "valueType should be DataType" 346 self.keyType = keyType AssertionError: keyType should be DataType

Answer 1

这个方法 get_json_object 成功了！

org.withColumn('inn',get_json_object(org.raw, '$.inn')).show(1)

Answer 2

您的代码存在几个问题：

$"raw" 与 Scala API 一起使用，在 Pyspark 中使用 col("raw") 或直接作为字符串 "raw"
当使用 StringType 或任何其他类型时，在 python 中您需要添加括号 StringType()
getItem 是一个 Column 方法，但您在字符串中调用它 ('searchcard'.getItem('inn'))

下面是使用 from_json 函数使用完整工作示例更正的同一代码：

import pyspark.sql.functions as F
from pyspark.sql.types import StringType, MapType

org = spark.createDataFrame([
    (123, '{"inn":"123", "prof": "tkie"}')
], ["id", "raw"])

org.withColumn(
    'raw',
    F.from_json("raw", MapType(StringType(), StringType()))
).select(
    'id',
    F.col('raw').getItem('inn').alias('inn')
).show()

#+---+---+
#| id|inn|
#+---+---+
#|123|123|
#+---+---+

从 spark table 中提取值 json 给出 SyntaxError 错误或 keyType should be DataType 错误

Extracting value from json from spark table gives SyntaxError error or keyType should be DataType error

python

json

apache-spark

pyspark

apache-spark-sql