PySpark объединяет поля в JSON с кадром данных

Я пытаюсь извлечь некоторые поля из строки JSONn в фрейм данных. Я могу добиться этого, поместив каждое поле в фрейм данных, а затем объединив все фреймы данных, как показано ниже. Но есть ли более простой способ сделать это? Потому что это всего лишь упрощенный пример, и мне нужно извлечь гораздо больше полей в моем проекте.

    from pyspark.sql import Row
    s = '{"job_id":"123","settings":{"task":[{"taskname":"task1"},{"taskname":"task2"}]}}'

    json_object = json.loads(s)

    # json_object

    job_id_l = [Row(job_id=json_object['job_id'])] 
    job_id_df = spark.createDataFrame(job_id_l) 
    # display(job_id_df)


    tasknames = []
    for t in json_object['settings']["task"]:
      tasknames.append(Row(taskname=t["taskname"]))

    tasknames_df = spark.createDataFrame(tasknames) 
    # display(tasknames_df)

    job_id_df.crossJoin(tasknames_df).display()

Результат:

    job_id  taskname
    123 task1
    123 task2

python json pyspark

05.07.2024 10:05

Почему в Python есть оператор "pass"?

Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.

Некоторые методы, о которых вы не знали, что они существуют в Python

Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...

Основы Python Часть I

Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?

LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа

Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:

Оптимизация кода с помощью тернарного оператора Python

И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это

Советы по эффективной веб-разработке с помощью Python

Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.

Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Обновление1:

Вам даже не нужно явно определять схему здесь, вместо этого вы можете просто использовать Schema_of_json следующим образом:

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, explode, schema_of_json, lit

spark = SparkSession.builder.getOrCreate()

s = '{"job_id":"123","settings":{"task":[{"taskname":"task1"},{"taskname":"task2"}]}}'

schema = schema_of_json(lit(s))

result_df = (
    spark.createDataFrame([s], "string")
    .select(from_json(col("value"), schema).alias("data"))
    .select("data.job_id", explode("data.settings.task.taskname").alias("taskname"))
)

result_df.show()

# +------+--------+
# |job_id|taskname|
# +------+--------+
# |   123|   task1|
# |   123|   task2|
# +------+--------+

Как вы упомянули, полей намного больше — это отнимет у вас часть работы.

Оригинальный ответ:

Более простой способ — отразить схему с помощью строки JSON s и использовать from_json следующим образом:

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, explode

spark = SparkSession.builder.getOrCreate()

s = '{"job_id":"123","settings":{"task":[{"taskname":"task1"},{"taskname":"task2"}]}}'

schema = "struct<job_id:string, settings:struct<task:array<struct<taskname:string>>>>"

result_df = (
    spark.createDataFrame([s], "string")
    .select(from_json(col("value"), schema).alias("data"))
    .select("data.job_id", explode("data.settings.task.taskname").alias("taskname"))
)

result_df.show()

# +------+--------+
# |job_id|taskname|
# +------+--------+
# |   123|   task1|
# |   123|   task2|
# +------+--------+

05.07.2024 11:38