парсинг столбца данных Pyspark в виде строки JSON со строкой массива в одном из столбцов

Question

Я пытаюсь прочитать файл JSON и выполнить синтаксический анализ 'jsonString' и базовых полей, включающих массив в фрейм данных pyspark.

Вот содержимое файла json.

[{"jsonString": "{\"uid\":\"value1\",\"adUsername\":\"value3\",\"courseCertifications\":[{\"uid\":\"value2\",\"courseType\":\"TRAINING\"},{\"uid\":\"TEST\",\"courseType\":\"TRAINING\"}],\"modifiedBy\":\"value4\"}","transactionId": "value5", "tableName": "X"},
 {"jsonString": "{\"uid\":\"value11\",\"adUsername\":\"value13\",\"modifiedBy\":\"value14\"}","transactionId": "value15", "tableName": "X1"},
 {"jsonString": "{\"uid\":\"value21\",\"adUsername\":\"value23\",\"modifiedBy\":\"value24\"}","transactionId": "value25", "tableName": "X2"}]

Я могу проанализировать содержимое строки 'jsonString' и выбрать необходимые столбцы, используя приведенную ниже логику

df = spark.read.json('path.json',multiLine=True)
df = df.withColumn('courseCertifications', explode(array(get_json_object(df['jsonString'],'$.courseCertifications'))))

Теперь моя конечная цель - проанализировать поле "courseType" из "courseCertifications" и создать одну строку для каждого экземпляра.

Я использую логику ниже, чтобы получить "courseType"

df = df.withColumn('new',get_json_object(df.courseCertifications, '$[*].courseType'))

Я могу получить содержимое "courseType", но в виде строки, как показано ниже

[Row(new=u'["TRAINING","TRAINING"]')]

Моя конечная цель - создать фрейм данных со столбцами.

• Мне нужно сохранить все строки и создать несколько строк, по одной на каждый экземпляр массива courseCertifications.uid / courseCertifications.courseType.

Answer 1

Элегантный способ решить ваш вопрос - создать схему строки json, а затем проанализировать ее, используя from_json function

import pyspark.sql.functions as f
from pyspark.shell import spark
from pyspark.sql.types import ArrayType, StringType, StructType, StructField

df = spark.read.json('your_path', multiLine=True)
schema = StructType([
    StructField('uid', StringType()),
    StructField('adUsername', StringType()),
    StructField('modifiedBy', StringType()),
    StructField('courseCertifications', ArrayType(
        StructType([
            StructField('uid', StringType()),
            StructField('courseType', StringType())
        ])
    ))
])

df = df \
    .withColumn('tmp', f.from_json(df.jsonString, schema)) \
    .withColumn('adUsername', f.col('tmp').adUsername) \
    .withColumn('uid', f.col('tmp').uid) \
    .withColumn('modifiedBy', f.col('tmp').modifiedBy) \
    .withColumn('tmp', f.explode(f.col('tmp').courseCertifications)) \
    .withColumn('course_uid', f.col('tmp').uid) \
    .withColumn('course_type', f.col('tmp').courseType) \
    .drop('jsonString', 'tmp')
df.show()

Вывод:

+-------------+------+----------+----------+----------+-----------+
|transactionId|uid   |adUsername|modifiedBy|course_uid|course_type|
+-------------+------+----------+----------+----------+-----------+
|value5       |value1|value3    |value4    |value2    |TRAINING   |
|value5       |value1|value3    |value4    |TEST      |TRAINING   |
+-------------+------+----------+----------+----------+-----------+