Apache.Spark.SQL 尝试将 JSON 转换为 Parquet 时遇到错误

【问题标题】：Running Into An Error With Apache.Spark.SQL Trying to Convert JSON to ParquetApache.Spark.SQL 尝试将 JSON 转换为 Parquet 时遇到错误
【发布时间】：2022-06-13 01:27:09
【问题描述】：

我正在尝试使用 apache.spark.sql 将 JSON 文件转换为 Parquet 文件，并且遇到了我认为读取 JSON 文件的问题。我附上了我的代码和 pom.xml 以及下面的错误消息。任何帮助将不胜感激，谢谢。

public class Main {
    public static void main(String[] args) {
        try {
            File file = new File("src/main/resources/HadoopResources");
            System.setProperty("hadoop.home.dir", file.getAbsolutePath());
            SparkSession spark = SparkSession
                .builder()
                .appName("Java Spark SQL basic example")
                .config("spark.master", "local")
                .getOrCreate();

            String path = "path\\JSONSAMPLEFILE.json";
            Dataset<Row> dr = spark.read().option("multiline", "true").json(path).cache();
            System.out.println(dr);
            System.out.println("AFter DR");

            dr.show();
            System.out.println("After Show");
            dr.write().parquet(JSONSAMPLEFILE.parquet ");
        } catch (Exception exception) {
            System.err.println(exception);
        }
        System.out.println("End");
    }
}

pom.xml 文件：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>org.example</groupId>
  <artifactId>jPFunc</artifactId>
  <version>1.0-SNAPSHOT</version>

  <properties>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>com.fasterxml.jackson.module</groupId>
      <artifactId>jackson-module-scala_2.11</artifactId>
      <version>2.8.8</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->

    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>3.3.3</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/com.datastax.spark/spark-cassandra-connector -->
    <dependency>
      <groupId>com.datastax.spark</groupId>
      <artifactId>spark-cassandra-connector_2.12</artifactId>
      <version>2.4.3</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.11</artifactId> <!-- matching Scala version -->
      <version>2.4.3</version> <!-- matching Spark Core version -->
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.11</artifactId>
      <version>2.4.3</version>
    </dependency>
    <dependency>
      <groupId>com.google.guava</groupId>
      <artifactId>guava</artifactId>
      <version>15.0</version>
    </dependency>
    <dependency>
      <groupId>io.netty</groupId>
      <artifactId>netty-all</artifactId>
      <version>4.1.77.Final</version>
    </dependency>
  </dependencies>
</project>

错误信息：

[_corrupt_record: string]
AFter DR
org.apache.spark.sql.AnalysisException: Since Spark 2.3, the queries from raw JSON/CSV files are disallowed when the
referenced columns only include the internal corrupt record column
(named _corrupt_record by default). For example:
spark.read.schema(schema).json(file).filter($"_corrupt_record".isNotNull).count()
and spark.read.schema(schema).json(file).select("_corrupt_record").show().
Instead, you can cache or save the parsed results and then send the same query.
For example, val df = spark.read.schema(schema).json(file).cache() and then
df.filter($"_corrupt_record".isNotNull).count().;
End

Process finished with exit code 0

【问题讨论】：

• 您在复制/粘贴时出现了语法错误。此行有误：dr.write().parquet(JSONSAMPLEFILE.parquet ");。

标签： java apache-spark apache-spark-sql