Spark SQL AnalysisException due to data type mismatch

If you run the following code, you will encounter AnalysisException with data type mismatch.

spark.sql("create table test_table(id int, data struct<name: string, age:int>)")
spark.sql("insert overwrite table test_table select 1, struct('joe', 15)")
val d = spark.read.table("test_table")

case class UserData(name: String, age: Int)
case class User(id: Int, data: UserData)
val u = Seq(User(1, UserData("joe", 15)), User(2, UserData("mary", 10))).toDF

val missing = {
  u.join(
    d,
    u("id") === d("id")
    && u("data")("name") === d("data")("name")
    && u("data")("age") === d("data")("age"),
    "outer"
  )
  .where( u("id").isNull || d("id").isNull)
}

// show this result
// +---+---------+----+----+                                                      
// |id |data     |id  |data|
// +---+---------+----+----+
// |2  |[mary,10]|null|null|
// +---+---------+----+----+
missing.show(false)

// Throws this error: org.apache.spark.sql.AnalysisException: cannot resolve '(`data` = test_table.`data`)'
// due to data type mismatch: differing types in '(`data` = test_table.`data`)'
// (struct<name:string,age:int> and struct<name:string,age:int>).;
val missing_2 = {
  u.join(d,
         u("id") === d("id") && u("data") === d("data"),
         "outer")
    .where(u("id").isNull || d("id").isNull)
}

Don’t be fooled by (struct<name:string,age:int> and struct<name:string,age:int>). The problem is caused by nullable, which is not shown in the error message. The DataFrame created from case classes has nullable=false for id and age because Scala Int cannot be null, while the SQL creates nullable fields. And if you compare a field with complex type (struct, array), Spark just thinks they are different as shown in missing_2. But if you compare field by field, there is no problem as shown in missing.

scala> u.schema.treeString
res4: String =
"root
 |-- id: integer (nullable = false)
 |-- data: struct (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- age: integer (nullable = false)
"

scala> d.schema.treeString
res5: String =
"root
 |-- id: integer (nullable = true)
 |-- data: struct (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- age: integer (nullable = true)
"

​

Make Spark read Teradata directly.

Spark SQL supports read JDBC resource, but the paragraph in the latest 2.0 document is not really helpful:

The JDBC driver class must be visible to the primordial class loader on the client session and on all executors. This is because Java’s DriverManager class does a security check that results in it ignoring all drivers not visible to the primordial class loader when one goes to open a connection. One convenient way to do this is to modify compute_classpath.sh on all worker nodes to include your driver JARs.

I wrote this blog to explain how to run against Teradata database:

• in spark-shell and YARN-cluster mode
• spark-submit and YARN-cluster mode

spark-shell and YARN mode

...
--conf spark.executor.extraClassPath=./tdgssconfig-15.10.00.22.jar:./terajdbc4-15.10.00.22.jar \
--driver-class-path $LIB_DIR/tdgssconfig-15.10.00.22.jar:$LIB_DIR/terajdbc4-15.10.00.22.jar \
...

• for executor.extraClassPath, the path is the current directory ./, which is where YARN starts the executor.
• for --driver-class-path, “$LIB_DIR” is the directory where those jdbc driver jars live, it is on the host where you run spark-shell.

spark-submit and YARN-cluster mode

...
--conf spark.executor.extraClassPath=./tdgssconfig-15.10.00.22.jar:./terajdbc4-15.10.00.22.jar \
--driver-class-path ./tdgssconfig-15.10.00.22.jar:./terajdbc4-15.10.00.22.jar \
--jars $LIB_DIR/tdgssconfig-15.10.00.22.jar,$LIB_DIR/terajdbc4-15.10.00.22.jar,<other jars>
...

Explanation

Two differences:

• In spark-shell, you don’t have to put Terdata jdbc driver jars in --jars, because --driver-class-path already does. If you put them in --jars, it doesn’t hurt. But in spark-submit, you have to add them in --jars.
• In spark-submit, --driver-class-path uses the current directory ./, not $LIB_IDR in spark-shell.

Lost? Here is why:

• When you run spark-submit with YARN-cluster mode , the Spark app driver actually runs in a YARN container, not where you type “spark-submit”.
  • --jars makes teradata jdbc driver jars copied to the container directory where the Spark app driver (YARN appMaster) runs.
  • --driver-class-path just declares you need teradata jdbc driver jars. Be the Spark app driver runs in a YARN container, this argument tells the spark app driver JVM where to find the jdbc jars. Because the jars are copied into the YARN container’s directory, to the JVM of spark app driver, the container directory is the current directory.
  • spark-submit actually cand not find the jdbc jars in ./ in --driver-class-path, so the jdbc jars won’t be copied to the YARN container. That is why you have to use --jars to tell spark where to find those jars and ship them to the container.
• In spark-shell, the spark app driver is started on the host where you type “spark-shell”. So you need to give the full path in --driver-class-path

​

Spark Cassandra Connector and DataFrame

When you write a DataFrame to a cassandra table, be careful to use SaveMode.Overwrite. In spark-cassandra-connector-1.6.0-M2, TRUNCATE $keyspace.$table will be called. See the code in CassandraSourceRelation.scala.

I did observe something weird when I use the following code to write a data frame to a cluster of Cassandra 2.1.8:

df.write
  .format(""org.apache.spark.sql.cassandra")
  .mode(SaveMode.Overwrite)
  .options(Map("table" -> table, "keyspace" -> keyspace))
  .save

After the scheduled spark job finishes, in CQLSH, the table is empty when running select * from keyspace.table limit 10. The same results if I change consistency level to QUORUM, and even ALL. It might take some time, then the query returns the results.

If I start the job manually from the command line, however, most of time the query can return the results.

If you check the CQL document for TRUNCATE, setting consistency level to ALL is required.

Note: The consistency level must be set to ALL prior to performing a TRUNCATE operation. All replicas must remove the data.

I don’t think the consistency level is changed before calling TRUNCATE $keyspace.$table in spark-cassandra-connector. The default consistency level is LOCAL_QUORUM. That might be the root cause.

​