SparkSQL DataFrame：使用缓存时 sql 查询不起作用

Question

我开始使用 spark 进行学习。我根据this document做了一个简单的程序。

我的程序从文件（在 HDFS 集群上）读取付款日志，将其传输到数据帧并在一些 sql 查询中使用此数据帧。我在两种情况下运行我的程序：使用和不使用 cache() 方法。我遇到了一个奇怪的问题，如下所述：

1. 不使用缓存（）：

我尝试运行一些查询，一切都很好。 （log_zw 是我的表名）

val num_records =  sqlContext.sql("select * from log_zw").count
val num_acc1 =  sqlContext.sql("select * from log_zw where ACN = 'acc1' ").count

1. 使用缓存()

我还使用了上面的两个查询。第一个查询返回了正确的值，但第二个是not，它返回了0。

但是，当我用另一种方法查询它时：

val num_acc1 = log_zw.filter(log_zw("ACN").contains("acc1")).count

它返回了正确的结果。

我对 Spark 和集群计算系统非常陌生，我不知道它为什么会这样工作。谁能给我解释一下这个问题，特别是使用sql查询和spark方法时的不同。

编辑：这是架构，非常简单。

root 
 |-- PRODUCT_ID: string (nullable = true) 
 |-- CHANNEL: string (nullable = true) 
 |-- ACN: string (nullable = true) 
 |-- AMOUNT_VND: double (nullable = false) 
 |-- TRANS_ID: string (nullable = true)

Edit2：这是我使用 cache() 时的代码：（我运行了一些查询，结果显示在代码中的 cmets 中）

// read tsv files
case class LogZW(
  PRODUCT_ID: String,
  PLATFORM: String,
  CHANNEL: String,
  ACN: String,
  AMOUNT_VND: Double,
  TRANS_ID: String)

def loadLog(filename: String): DataFrame = {
  sc.textFile(filename).map(line => line.split("\t")).map(p =>
  LogZW(p(1), p(3), p(4), p(5), p(9).toDouble, p(10).substring(0,8))).toDF()
}

// generate schema
val schemaString = "PRODUCT_ID PLATFORM CHANNEL ACN AMOUNT_VND TRANS_ID"
val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))

// read all files
val HDFSFolder = "hdfs://master:54310/user/lqthang/data/*"
val log = loadLog(HDFSFolder)

// register table
log.registerTempTable("log")
log.show()

// select a subset of log table
val log_zw =  sqlContext.sql("select PRODUCT_ID, CHANNEL, ACN, AMOUNT_VND, TRANS_ID from log where PLATFORM = 'zingwallet' and CHANNEL not in ('CBZINGDEAL', 'VNPT') and PRODUCT_ID not in ('ZingCredit', 'zingcreditdbg') ")

// register new table
log_zw.show()
log_zw.registerTempTable("log_zw")

// cache table
log_zw.cache()

// this query returns incorrect value!!
val num_acc1 =  sqlContext.sql("select * from log_zw where ACN = 'acc1' ").count

// this query returns correct value!
val num_acc2 =  sqlContext.sql("select * from log_zw where trim(ACN) = 'acc1' ").count

// uncache data and try another query
log_zw.unpersist()

// this query also returns the correct value!!!
val num_acc2 =  sqlContext.sql("select * from log_zw where ACN = 'acc1' ").count

Edit3：我尝试向log dataframe 添加另一个cache() 方法：

// register table
log.registerTempTable("log")
log.show()
log.cache()

下面的代码与上面相同（带有log_zw.cache()）。所以重要的结果是：

// this query returns the CORRECT value!!
val num_acc1 =  sqlContext.sql("select * from log_zw where ACN = 'acc1' ").count

Answer 1

我们没有太多关于数据是什么的详细信息，但我注意到你的两个代码部分做不同的事情。

在第一个中，您执行ACN = 'acc1'，但在第二个中您检查 ACN 是否包含'acc1'。

因此，如果 ACN 为“acc1”、“acc1”或“acc1”，则第二位（带有过滤器）将匹配

换句话说，我敢打赌，如果您在 SQL 查询中添加修剪，您会得到不同的结果。

所以试试这个：
val num_records = sqlContext.sql("select * from log_zw").count val num_acc1 = sqlContext.sql("select * from log_zw where trim(ACN) = 'acc1' ").count