.csv 不是 Select Hive Query 上的 SequenceFile 错误

Posted 2023-03-23

技术标签:

【中文标题】.csv 不是 Select Hive Query 上的 SequenceFile 错误【英文标题】：.csv not a SequenceFile error on Select Hive Query 【发布时间】：2021-09-08 23:41:53 【问题描述】：

我是 Spark 和 Scala 的新手 ;)

代码总结：

从 CSV 文件中读取数据 --> 在 2 个文件上创建简单的内部连接 --> 将数据写入 Hive 表 --> 在集群上提交作业

请您帮忙找出问题所在。代码并不复杂。该作业在集群上执行良好。因此，当我尝试可视化写在 hive 表上的数据时，我遇到了问题。

hive> select * from Customers limit 10;

失败并出现异常 java.io.IOException:java.io.IOException: hdfs://m01.itversity.com:9000/user/itv000666/warehouse/updatedcustomers.db/customers/part-00000-348a54cf -aa0c-45b4-ac49-3a881ae39702_00000.c000 .csv 不是序列文件

object LapeyreSparkDemo extends App 
  
  //Getting spark ready
  val sparkConf = new SparkConf()
  sparkConf.set("spark.app.name","Spark for Lapeyre")
  
  //Creating Spark Session
  val spark = SparkSession.builder()
                          .config(sparkConf)
                          .enableHiveSupport()
                          .config("spark.sql.warehouse.dir","/user/itv000666/warehouse")
                          .getOrCreate()                       
  Logger.getLogger(getClass.getName).info("Spark Session Created Successfully")
  
  //Reading
  Logger.getLogger(getClass.getName).info("Data loading in DF started")
  val ordersSchema = "orderid Int, customerName String, orderDate String, custId Int, orderStatus 
  String, age String, amount Int" 
  val orders2019Df = spark.read
  .format("csv")
  .option("header",true)
  .schema(ordersSchema)
  .option("path","/user/itv0006666/lapeyrePoc/orders2019.csv")
  .load
  val newOrder = orders2019Df.withColumnRenamed("custId", "oldCustId")
                             .withColumnRenamed("customername","oldCustomerName")
   
  val orders2020Df = spark.read
  .format("csv")
  .option("header",true)
  .schema(ordersSchema)
  .option("path","/user/itv000666/lapeyrePoc/orders2020.csv")
  .load
  
  Logger.getLogger(getClass.getName).info("Data loading in DF complete")
  
  //processing
  Logger.getLogger(getClass.getName).info("Processing Started")
  val joinCondition = newOrder.col("oldCustId") === orders2020Df.col("custId")
  val joinType = "inner"
  val joinData = newOrder.join(orders2020Df, joinCondition, joinType)
                             .select("custId","customername")
  
  //Writing
  
  spark.sql("create database if not exists updatedCustomers")
                  
  joinData.write
  .format("csv")
  .mode(SaveMode.Overwrite)
  .bucketBy(4, "custId")
  .sortBy("custId")
  .saveAsTable("updatedCustomers.Customers")
                        
  //Stopping Spark Session
  spark.stop()

如果需要更多信息，请告诉我。提前致谢。

【问题讨论】：

您可以将describe extended 的结果添加到问题中吗？也许会有所帮助。 DESCRIBE [DATABASE] [FORMATTED|EXTENDED] object_name 【参考方案1】：

这是罪魁祸首