Spark DataFrame解析json字符串

栏目: 服务器 · 发布时间: 7年前

内容简介:近期在做Spark Streaming方面的测试,从Kafka中实时取数据。此时接收到的数据是一段json数组形式的字符串,那么就需要将其解析为DataFrame的行数据,以方便进行实时运算。下面的代码示例演示了如何实现这个功能,因为比较简单,就多作说明了。假设初始的字符串是:[{"a":1,"b":2},{"a":3,"b":4},{"a":5,"b":6},{"a":7,"b":8}],最终将其解析为下面的DataFrame:df.show()只显示了struct的字段值,没有显示字段名称,并且用[]

近期在做Spark Streaming方面的测试,从Kafka中实时取数据。此时接收到的数据是一段json数组形式的字符串,那么就需要将其解析为DataFrame的行数据,以方便进行实时运算。下面的代码示例演示了如何实现这个功能,因为比较简单,就多作说明了。

假设初始的字符串是:[{"a":1,"b":2},{"a":3,"b":4},{"a":5,"b":6},{"a":7,"b":8}],最终将其解析为下面的DataFrame:

a b
1 2
3 4
5 6
7 8
# coding=utf-8
from pyspark.sql import SparkSession
from pyspark.sql import types as T
from pyspark.sql import functions as F

spark = SparkSession.builder\
    .master("local[*]")\
    .appName("test.dataframe")\
    .getOrCreate()

# 第1步,加载数据,默认为字符串类型的单列,列名为value
data = ['[{"a":1,"b":2},{"a":3,"b":4},{"a":5,"b":6},{"a":7,"b":8}]']
df = spark.createDataFrame(data, T.StringType())
df.printSchema()
df.show()

schema = T.ArrayType(T.StructType([
    T.StructField("a", T.IntegerType()),
    T.StructField("b", T.IntegerType())
]))

# 第2步,将列转为数组类型
df = df.select(F.from_json(df["value"], schema).alias("json"))
df.printSchema()
df.show()

# 第3步,将列转为Struct类型
df = df.select(F.explode(df["json"]).alias("col"));
df.printSchema()
df.show()

# 第4步,对Struct进行拆分
col = df["col"]
df = df.withColumn("a", col["a"]) \
    .withColumn("b", col["b"]) \
    .drop("col")
df.printSchema()
df.show()

print "success"

df.show()只显示了struct的字段值,没有显示字段名称,并且用[],而不是{}来代表Struct,所以看上去像是一个数组

其执行结果如下:

# $SPARK_HOME/bin/spark-submit /data/pyjobs/sample/dataframe-json.py
root
 |-- value: string (nullable = true)

+--------------------+
|               value|
+--------------------+
|[{"a":1,"b":2},{"...|
+--------------------+

root
 |-- json: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: integer (nullable = true)
 |    |    |-- b: integer (nullable = true)

+--------------------+
|                json|
+--------------------+
|[[1, 2], [3, 4], ...|
+--------------------+

root
 |-- col: struct (nullable = true)
 |    |-- a: integer (nullable = true)
 |    |-- b: integer (nullable = true)

+------+
|   col|
+------+
|[1, 2]|
|[3, 4]|
|[5, 6]|
|[7, 8]|
+------+

root
 |-- a: integer (nullable = true)
 |-- b: integer (nullable = true)

+---+---+
|  a|  b|
+---+---+
|  1|  2|
|  3|  4|
|  5|  6|
|  7|  8|
+---+---+

success

感谢阅读,希望这篇文章能给你带来帮助!

Spark DataFrame解析json字符串


以上所述就是小编给大家介绍的《Spark DataFrame解析json字符串》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

软件的奥秘

软件的奥秘

[美] V. Anton Spraul / 解福祥 / 人们邮电出版社 / 2017-9-1 / 49

软件已经成为人们日常生活与工作中常见的辅助工具,但是对于软件的工作原理,很多人却不是非常了解。 本书对软件的工作原理进行了解析,让读者对常用软件的工作原理有一个大致的了解。内容涉及数据如何加密、密码如何使用和保护、如何创建计算机图像、如何压缩和存储视频、如何搜索数据、程序如何解决同样的问题而不会引发冲突以及如何找出最佳路径等方面。 本书适合从事软件开发工作的专业技术人员,以及对软件工作......一起来看看 《软件的奥秘》 这本书的介绍吧!

CSS 压缩/解压工具
CSS 压缩/解压工具

在线压缩/解压 CSS 代码

html转js在线工具
html转js在线工具

html转js在线工具

RGB CMYK 转换工具
RGB CMYK 转换工具

RGB CMYK 互转工具