来源 https://stackoverflow.com/questions/41107835/pyspark-parse-a-column-of-json-strings

这是一个很有效的解决方法。

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(
                    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

来源https://stackoverflow.com/questions/41107835/pyspark-parse-a-column-of-json-strings这是一个很有效的解决方法。def parseJSONCols(df, *cols, sanitize=True): """Auto infer the schema of a json column and pa... def to_ json 1(df,orient='split'): return df.to_ json (orient = orient, force_ascii = False) def to_ json 2(df,o... 法1： dataFrame 数据写入hive表 def log2Hive(): log=hiveContext.create DataFrame ([{"dt":dt,"types":types,"message":msg,"currtime":currTime}]).coalesce(1)#types: "INFO" ,"ERROR" log.write.mode("ap 自定义函数的重点在于定义返回值类型的数据 格式 ，其数据类型基本都是从from pyspark .sql.types import * 导入，常用的包括： - StructType()：结构体 - StructField()：结构体 中 的元素 - LongT... 环境：zeppelin 中 的spark 2.1 的notebook提交的代码 pyspark 读 json dataframe = spark.read.format(“ json ”).load("/tmp/testhdfsfile") #路径是hdfs上的 注意 json 文件 中 ，一条记录是一条 json ，不能换行， 格式 如下： {“row”:“1”,“field1”:“valu...