临时表上的多个SQL失败

Question

Spark Version: 1.6.2.   

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中包含的所有目录，并在第二次保存数据文件时使用以下

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存储到其他文件中