在Spark中,我们遇到了同样的问题。我们正在使用以下内容:
from pyspark.sql.functions import *
@udf
def concatenated_json_to_array(text):
final = "["
separator = ""
for part in text.split("}{"):
final += separator + part
separator = "}{" if re.search(r':\s*"([^"]|(\\"))*$', final) else "},{"
return final + "]"
def read_concatenated_json(path, schema):
return (spark.read
.option("lineSep", None)
.text(path)
.withColumn("value", concatenated_json_to_array("value"))
.withColumn("value", from_json("value", schema))
.withColumn("value", explode("value"))
.select("value.*"))
它的工作原理如下:
- 每个文件将数据读取为一个字符串(没有分隔符!)
- 使用UDF引入JSON数组,并通过引入逗号拆分JSON对象。注意:要小心不要破坏其中任何带有
}{
的字符串!
- 使用模式解析JSON到DataFrame字段中。
- 将数组展开为单独的行。
- 将值对象扩展到列中。
像这样使用它:
from pyspark.sql.types import *
schema = ArrayType(
StructType([
StructField("type", StringType(), True),
StructField("value", StructType([
StructField("id", IntegerType(), True),
StructField("joke", StringType(), True),
StructField("categories", ArrayType(StringType()), True)
]), True)
])
)
path = '/mnt/my_bucket_name/messages/*/*/*/*/'
df = read_concatenated_json(path, schema)
我在这里写了更多的细节和注意事项:使用Spark解析来自S3(Kinesis)的JSON数据。不要仅仅通过}{
进行分割,因为它可能会破坏你的字符串数据!例如:{ "line": "a\"r}{t" }
。