Pyspark: проанализировать столбец строк json

Ответ 1

Преобразование DataFrame с json-строками в структурированный dataframe is'a на самом деле довольно прост в искровом случае, если раньше вы конвертировали dataframe в RDD строк (см. http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets)

Например:

>>> new_df = sql_context.read.json(df.rdd.map(lambda r: r.json))
>>> new_df.printSchema()
root
 |-- body: struct (nullable = true)
 |    |-- id: long (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- sub_json: struct (nullable = true)
 |    |    |-- id: long (nullable = true)
 |    |    |-- sub_sub_json: struct (nullable = true)
 |    |    |    |-- col1: long (nullable = true)
 |    |    |    |-- col2: string (nullable = true)
 |-- header: struct (nullable = true)
 |    |-- foo: string (nullable = true)
 |    |-- id: long (nullable = true)

Ответ 2

Для Spark 2. 1+ вы можете использовать from_json который позволяет сохранять другие столбцы не-json в кадре данных следующим образом:

from pyspark.sql.functions import from_json, col
json_schema = spark.read.json(df.rdd.map(lambda row: row.json)).schema
df.withColumn('json', from_json(col('json'), json_schema))

Вы позволяете Spark получить схему строкового столбца json. Тогда столбец df.json больше не является StringType, а правильно декодированная структура json, то есть вложенный StrucType и все остальные столбцы df, сохраняются как есть.

Вы можете получить доступ к содержимому json следующим образом:

df.select(col('json.header').alias('header'))

Ответ 3

Существующие ответы не работают, если ваш JSON отличается от идеального/традиционного формата. Например, вывод схемы на основе RDD ожидает JSON в фигурных скобках {} и предоставит неверную схему (в результате чего будут null значения), если, например, ваши данные выглядят следующим образом:

[
  {
    "a": 1.0,
    "b": 1
  },
  {
    "a": 0.0,
    "b": 2
  }
]

Я написал функцию, чтобы обойти эту проблему, очистив JSON так, чтобы он находился в другом объекте JSON:

def parseJSONCols(df, *cols, sanitize=True):
    """Auto infer the schema of a json column and parse into a struct.

    rdd-based schema inference works if you have well-formatted JSON,
    like ''{"key": "value", ...}'', but breaks if your 'JSON' is just a
    string (''"data"'') or is an array (''[1, 2, 3]''). In those cases you
    can fix everything by wrapping the data in another JSON object
    (''{"key": [1, 2, 3]}''). The ''sanitize'' option (default True)
    automatically performs the wrapping and unwrapping.

    The schema inference is based on this
    'SO Post <https://stackoverflow.com/a/45880574)/>'_.

    Parameters
    ----------
    df : pyspark dataframe
        Dataframe containing the JSON cols.
    *cols : string(s)
        Names of the columns containing JSON.
    sanitize : boolean
        Flag indicating whether you'd like to sanitize your records
        by wrapping and unwrapping them in another JSON object layer.

    Returns
    -------
    pyspark dataframe
        A dataframe with the decoded columns.
    """
    res = df
    for i in cols:

        # sanitize if requested.
        if sanitize:
            res = (
                res.withColumn(
                    i,
                    psf.concat(psf.lit('{"data": '), i, psf.lit('}'))
                )
            )
        # infer schema and apply it
        schema = spark.read.json(res.rdd.map(lambda x: x[i])).schema
        res = res.withColumn(i, psf.from_json(psf.col(i), schema))

        # unpack the wrapped object if needed
        if sanitize:
            res = res.withColumn(i, psf.col(i).data)
    return res

Примечание: psf= pyspark.sql.functions.

Ответ 4

Здесь краткая (искристая SQL) версия функции @nolan-conaway parseJSONCols.

SELECT 
explode(
    from_json(
        concat('{"data":', 
               '[{"a": 1.0,"b": 1},{"a": 0.0,"b": 2}]', 
               '}'), 
        'data array<struct<a:DOUBLE, b:INT>>'
    ).data) as data;

PS. Я также добавил функцию разнесения: P

Вам нужно знать некоторые типы HIVE SQL