临时表上的多个 SQL 失败

Question

Spark Version: 1.6.2.   

我注册了一个临时表，数据源是HDFS，查询了两次。

然后作业因以下错误而失败：

错误 ApplicationMaster：用户类抛出异常：
java.io.IOException：不是文件：hdfs://my_server:8020/2017/01/01
java.io.IOException：不是文件：hdfs://my_server:8020/2017/01/01 在 org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:322) 在 org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:242) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:240) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:240)

棘手的部分是，如果只运行一个查询，作业就会成功。
我是否以错误的方式使用 Spark SQL，或者这是故意的？

这是我的代码的样子：

val rdd = sc.textFile("hdfs://my_server:8020/2017/*/*/*")
val table = sqlc.read.json(rdd).cache()

table.registerTempTable("my_table")

sql("""
    | SELECT contentsId,
    |   SUM(CASE WHEN gender = 'M' then 1 else 0 end)
    | FROM my_table
    | GROUP BY contentsId
  """.stripMargin)
  .write.format("com.databricks.spark.csv")
  .save("hdfs://my_server:8020/gender.csv")

sql("""
    | SELECT contentsId,
    |   SUM(CASE WHEN age > 0 AND age < 20 then 1 else 0 end),
    |   SUM(CASE WHEN age >= 20 AND age < 30 then 1 else 0 end)
    | FROM my_table
    | GROUP BY contentsId
  """.stripMargin)
  .write.format("com.databricks.spark.csv")
  .save("hdfs://my_server:8020/age.csv")

提前致谢！

Answer 1

我认为您可以尝试仅对此类文件应用过滤器。

val filesRDD = rdd.filter{path => (new java.io.File(path).isFile)}

这将删除 RDD 中包含的所有目录 并且第二次保存 DataFrame 使用这个

sql("""
    | SELECT contentsId,
    |   SUM(CASE WHEN age > 0 AND age < 20 then 1 else 0 end),
    |   SUM(CASE WHEN age >= 20 AND age < 30 then 1 else 0 end)
    | FROM my_table
    | GROUP BY contentsId
  """.stripMargin)
  .write.format("com.databricks.spark.csv")
  .mode("append")
  .save("hdfs://my_server:8020/gender.csv")

如果存储值相同或尝试将第二个 DataFrame 存储到其他文件中