I noticed this while trying to load and clean a csv file with some rows not complying to the schema. The following test case illustrates what I'm seeing.
test.csv:
ROWNUM,SOME_INT,SOME_STRING
1,12,aaa
2,NA,bbb
3,32,ccc
The second row won't comply with the schema because of that NA.
I loaded the file like this:
Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_66)
Type in expressions to have them evaluated.
Type :help for more information.
scala> import org.apache.spark.sql.types.{
| DataTypes, StructType, StructField, StringType, IntegerType}
import org.apache.spark.sql.types.{DataTypes, StructType, StructField, StringType, IntegerType, DecimalType}
scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row
scala> val dfSchema = StructType(
| StructField("ROWNUM", IntegerType, true)
| :: StructField("SOME_INT", IntegerType, false)
| :: StructField("SOME_STRING", StringType, false)
| :: Nil)
dfSchema: org.apache.spark.sql.types.StructType = StructType(StructField(ROWNUM,IntegerType,true), StructField(SOME_INT,IntegerType,false), StructField(SOME_STRING,StringType,false))
scala> val df = spark.read.format("csv").option("header", "true").schema(dfSchema).load("/Users/user/test.csv")
df: org.apache.spark.sql.DataFrame = [ROWNUM: int, SOME_INT: int ... 1 more field]
scala> df.head(3)
res1: Array[org.apache.spark.sql.Row] = Array([1,12,aaa], [null,null,null], [3,32,ccc])
The second row has all nulls because "NA" is not an integer - fair enough. But when I tried to detect / isolate the null rows I saw behaviour I didn't expect:
scala> val dfNullRows = df.filter(df.col("ROWNUM").isNull) // expecting 1 row
dfNullRows: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [ROWNUM: int, SOME_INT: int ... 1 more field]
scala> dfNullRows.count()
res2: Long = 0
scala> dfNullRows.show()
+------+--------+-----------+
|ROWNUM|SOME_INT|SOME_STRING|
+------+--------+-----------+
| null| null| null|
+------+--------+-----------+
scala> val dfNonNullRows = df.filter(df.col("ROWNUM").isNotNull) // expecting 2 rows
dfNonNullRows: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [ROWNUM: int, SOME_INT: int ... 1 more field]
scala> dfNonNullRows.count()
res4: Long = 3
scala> dfNonNullRows.show()
+------+--------+-----------+
|ROWNUM|SOME_INT|SOME_STRING|
+------+--------+-----------+
| 1| 12| aaa|
| 3| 32| ccc|
+------+--------+-----------+
scala> val dfRow2 = df.filter(df.col("ROWNUM") === 2) // expecting empty DataFrame
dfRow2: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [ROWNUM: int, SOME_INT: int ... 1 more field]
scala> dfRow2.count()
res6: Long = 1
scala> dfRow2.show()
+------+--------+-----------+
|ROWNUM|SOME_INT|SOME_STRING|
+------+--------+-----------+
+------+--------+-----------+
show() is giving me what I expect in all three cases above, but count() does not seem to be consistent with that.
Am I making some invalid assumption about what's happening here? It's almost as if count() is operating on a version of the dataframe(s) without the specified schema.