Spark SQL 读取已转义双引号的 JSON 文件
Posted
技术标签:
【中文标题】Spark SQL 读取已转义双引号的 JSON 文件【英文标题】:Spark SQL read a JSON file which has already escaped double quote 【发布时间】:2017-11-07 01:11:06 【问题描述】:我有一个简单的 Spark 程序,它读取 JSON 文件并发出 CSV 文件。在 JSON 文件中,数据被转义为双引号。 spark 程序无法将该行读取为有效的 JSON 字符串。
输入.json
\"key\" : \"k1\", \"value1\": \"Good String\", \"value2\": \"Good String\"
input_1.json
"\"key\" : \"k1\", \"value1\": \"Good String\", \"value2\": \"Good String\""
output.csv - 数据作为损坏记录返回
_corrupt_record,key,value1,value2
"\\"key\\\" : \\\"k1\\\", \\\"value1\\\": \\\"Good String\\\", \\\"value2\\\": \\\"Good String\\\"",,,
expected.csv
,k1,Good String,Good String
请看下面的主要代码并指教
public static void main(String[] args)
SparkSession sparkSession = SparkSession.builder()
.appName(TestSpark.class.getName()).master("local[1]").getOrCreate();
SparkContext context = sparkSession.sparkContext();
context.setLogLevel("ERROR");
SQLContext sqlCtx = sparkSession.sqlContext();
List<StructField> kvFields = new ArrayList<>();
kvFields.add(DataTypes.createStructField("_corrupt_record", DataTypes.StringType, true));
kvFields.add(DataTypes.createStructField("key", DataTypes.StringType, true));
kvFields.add(DataTypes.createStructField("value1", DataTypes.StringType, true));
kvFields.add(DataTypes.createStructField("value2", DataTypes.StringType, true));
StructType employeeSchema = DataTypes.createStructType(kvFields);
Dataset<Row> dataset = sparkSession.read()
.option("inferSchema", false)
.format("json")
.schema(employeeSchema)
.load("D:\\dev\\workspace\\java\\simple-kafka\\key_value.json");
dataset.createOrReplaceTempView("sourceView");
sqlCtx.sql("select * from sourceView")
.write()
.option("header", true)
.format("csv")
.save("D:\\dev\\workspace\\java\\simple-kafka\\output\\" + UUID.randomUUID().toString());
sparkSession.close();
【问题讨论】:
【参考方案1】:您需要阅读 as 文本并手动解析。我不将 Spark 与 Java 一起使用,但这里是 Scala 等价物,您可以将其用作伪代码:
val rdd: RDD[MyClass] = sc.textFile(path)
.map line =>
val json = ... // turn line into valid json
Try(parse(json))
.recover
case NonFatal(ex) => // handle parse error
.map jvalue =>
// Convert the JSON into a case class
.get
val ds: Dataset[MyClass] = spark.createDataset(rdd)
【讨论】:
以上是关于Spark SQL 读取已转义双引号的 JSON 文件的主要内容,如果未能解决你的问题,请参考以下文章