如何使用PYSPARK处理关系数据库中的JSON字段？

Question

我正在尝试处理PostgreSQL数据库中的JSON列。我可以使用以下方式连接到数据库：

import os
import pyspark
import findspark
from pyspark import SparkContext
from pyspark.sql import SQLContext
findspark.init(os.environ['SPARK_HOME'])

# DB credentials
user = os.environ['EVENTS_DEV_UID']
password = os.environ['EVENTS_DEV_PWD']
host = os.environ['EVENTS_DEV_HOST']
port = os.environ['EVENTS_DEV_PORT']
db = os.environ['EVENTS_DEV_DBNAME']

# Initiate spark session
sc = SparkContext()
spark = SQLContext(sc)

# Set properties
properties = {"user": user, "password": password, "driver": "org.postgresql.Driver"}

# Load data
df = spark.read.jdbc(
    url = 'jdbc:postgresql://' + host + ':' + port + '/' + db,
    table = 'events',
    properties = properties)

问题始于铸造JSON字段。Spark不能识别params的结构格式。当我打印模式时：

df.printSchema()

root
|--time： time（nullable=true）
|--name： string（nullable=true）
|--params： string（nullable=true）

当我尝试将字符串强制转换为struct时：

df = df.withColumn('params', df.params.cast('struct'))

我遇到以下错误：

ParseException: '\nDataType struct is not supported.(line 1, pos 0)\n\n== SQL ==\nstruct\n^^^\n'

我想问题在于转义字符。有人知道如何进行吗？

Answer 1

结构不是有效的转换类型。您可以使用python的json.loads函数定义自己的UDF。让我们从一个示例数据帧开始：

df = sc.parallelize([[1, "a", "{\"a\":1, \"b\":2}"], [2, "b", "{\"a\":3, \"b\":4}"]])\
    .toDF(["col1", "col2", "json_col"])
df.show()

    +----+----+--------------+
    |col1|col2|      json_col|
    +----+----+--------------+
    |   1|   a|{"a":1, "b":2}|
    |   2|   b|{"a":3, "b":4}|
    +----+----+--------------+

然后，输出StructType将具有模式：

from pyspark.sql.types import IntegerType, StructField, StructType
schema = StructType([StructField("a", IntegerType()), StructField("b", IntegerType())])

不能将StringType强制转换为StructType，因此UDF：

import pyspark.sql.functions as psf
import json
json_load = psf.udf(json.loads, schema)

现在我们可以处理json_col：

df_parsed = df.withColumn("parsed_json", json_load("json_col"))
df_parsed.show()
df_parsed.printSchema()

    +----+----+--------------+-----------+
    |col1|col2|      json_col|parsed_json|
    +----+----+--------------+-----------+
    |   1|   a|{"a":1, "b":2}|      [1,2]|
    |   2|   b|{"a":3, "b":4}|      [3,4]|
    +----+----+--------------+-----------+

    root
     |-- col1: long (nullable = true)
     |-- col2: string (nullable = true)
     |-- json_col: string (nullable = true)
     |-- parsed_json: struct (nullable = true)
     |    |-- a: integer (nullable = true)
     |    |-- b: integer (nullable = true)

您还可以尝试在加载数据帧时直接传递模式。