不是一个Parquet文件。尾部的预期魔术数[80，65，82，49]，但发现[110，111，13，10]

Question

所以我试图加载推断自定义模式的csv文件，但是每次我遇到以下错误时：

不是一个Parquet文件。尾部的预期魔术数[80，65，82，49]，但发现[110，111，13，10]

这就是我的程序和csv文件条目的样子，

年龄;工作;婚姻;教育;默认;平衡;住房;贷款;联系方式;天;月;任期;活动;周日;以前;结果; y 58;管理;已婚;三级;否; 2143;是;否;未知; 5; may; 261; 1; -1; 0;未知;否44;技术员;单身;中学;否; 29;是;否;未知; 5;可能; 151; 1; -1; 0;未知;否; 33;企业家;已婚;中学;否; 2;是;是;未知; 5;可能; 76; 1; -1; 0;未知;否

我的代码：

$ spark-shell --packages com.databricks：spark-csv_2.10：1.5.0

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import org.apache.spark.sql.types._
import org.apache.spark.sql.SQLContext   
import sqlContext.implicits._    
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

val bankSchema = StructType(Array(
  StructField("age", IntegerType, true),
  StructField("job", StringType, true),
  StructField("marital", StringType, true),
  StructField("education", StringType, true),
  StructField("default", StringType, true),
  StructField("balance", IntegerType, true),
  StructField("housing", StringType, true),
  StructField("loan", StringType, true),
  StructField("contact", StringType, true),
  StructField("day", IntegerType, true),
  StructField("month", StringType, true),
  StructField("duration", IntegerType, true),
  StructField("campaign", IntegerType, true),
  StructField("pdays", IntegerType, true),
  StructField("previous", IntegerType, true),
  StructField("poutcome", StringType, true),
  StructField("y", StringType, true)))


 val df = sqlContext.
  read.
  schema(bankSchema).
  option("header", "true").
  option("delimiter", ";").
  load("/user/amit.kudnaver_gmail/hadoop/project_bank/Project_Bank.csv").toDF()

  df.registerTempTable("people")
  df.printSchema()
  val distinctage = sqlContext.sql("select distinct age from people")

关于为什么在推送正确的架构后为什么无法使用csv文件的任何建议。预先感谢您的建议。

谢谢阿米特·K

Answer 1

这里的问题是数据框架在处理它时需要 Parquet 文件。为了处理 CSV 中的数据。在这里你可以做什么。

首先，从数据中删除标题行。

58;management;married;tertiary;no;2143;yes;no;unknown;5;may;261;1;-1;0;unknown;no
44;technician;single;secondary;no;29;yes;no;unknown;5;may;151;1;-1;0;unknown;no
33;entrepreneur;married;secondary;no;2;yes;yes;unknown;5;may;76;1;-1;0;unknown;no

接下来我们编写如下代码来读取数据。

创建案例类

case class BankSchema(age: Int, job: String, marital:String, education:String, default:String, balance:Int, housing:String, loan:String, contact:String, day:Int, month:String, duration:Int, campaign:Int, pdays:Int, previous:Int, poutcome:String, y:String)

从HDFS读取数据并解析

val bankData = sc.textFile("/user/myuser/Project_Bank.csv").map(_.split(";")).map(p => BankSchema(p(0).toInt, p(1), p(2),p(3),p(4), p(5).toInt, p(6), p(7), p(8), p(9).toInt, p(10), p(11).toInt, p(12).toInt, p(13).toInt, p(14).toInt, p(15), p(16))).toDF()

然后注册表并执行查询。

bankData.registerTempTable("bankData")
val distinctage = sqlContext.sql("select distinct age from bankData")

输出如下所示

+---+
|age|
+---+
| 33|
| 44|
| 58|
+---+