I see -- they are the same in design but the difference comes from partitioned Hive tables: when the RDD is generated by querying an external Hive metastore, the partition is appended as part of the row, and shows up as part of the schema. Can you shed some light on why this is a problem:
last2HourRdd.first <-- works ok last2HourRdd.cache() last2HourRdd.first <-- does not work The first call shows K+1 columns (and so does print schema, where K columns are from the backing parquet files and the K+1st is the partition inlined. My impression is that the second call to .first would just force the cache() call and dump out that RDD to disk (with all of it's K+1 columns and store the schema info, again with K+1 columns), and then just return a single entry. I am not sure why the fact that Hive metastore exposes an extra column over the raw parquet file is a problem since it does so both on the schema and in the data: last2HourRdd.schema.fields.length reports K+1, and so does last2HourRdd.first.length. I also tried calling sqlContext.applySchema(last2HourRdd,parquetFile.schema) before caching but it does not fix the issue. The only workaround I've come up with so far is to replace select * with a select <list_of_columns>. But I'd love to understand a little better why the cache call trips this scenario On Wed, Dec 10, 2014 at 3:50 PM, Michael Armbrust <mich...@databricks.com> wrote: > Have you checked to make sure the schema in the metastore matches the > schema in the parquet file? One way to test would be to just use > sqlContext.parquetFile(...) which infers the schema from the file instead > of using the metastore. > > On Wed, Dec 10, 2014 at 12:46 PM, Yana Kadiyska <yana.kadiy...@gmail.com> > wrote: > >> >> Hi folks, wondering if anyone has thoughts. Trying to create something >> akin to a materialized view (sqlContext is a HiveContext connected to >> external metastore): >> >> >> val last2HourRdd = sqlContext.sql(s"select * from mytable") >> //last2HourRdd.first prints out a org.apache.spark.sql.Row = [...] with >> valid data >> >> last2HourRdd.cache() >> //last2HourRdd.first now fails in an executor with the following: >> >> In the driver: >> >> 14/12/10 20:24:01 INFO TaskSetManager: Starting task 0.1 in stage 25.0 (TID >> 35, iphere, NODE_LOCAL, 2170 bytes) >> 14/12/10 20:24:01 INFO TaskSetManager: Lost task 0.1 in stage 25.0 (TID 35) >> on executor iphere: java.lang.ClassCastException (null) [duplicate 1] >> >> >> >> >> And in executor: >> >> 14/12/10 19:56:57 ERROR Executor: Exception in task 0.1 in stage 20.0 (TID >> 27) >> java.lang.ClassCastException: java.lang.String cannot be cast to >> java.lang.Integer >> at scala.runtime.BoxesRunTime.unboxToInt(BoxesRunTime.java:106) >> at >> org.apache.spark.sql.catalyst.expressions.MutableInt.update(SpecificMutableRow.scala:73) >> at >> org.apache.spark.sql.catalyst.expressions.SpecificMutableRow.update(SpecificMutableRow.scala:231) >> at >> org.apache.spark.sql.catalyst.expressions.SpecificMutableRow.setString(SpecificMutableRow.scala:236) >> at org.apache.spark.sql.columnar.STRING$.setField(ColumnType.scala:328) >> at org.apache.spark.sql.columnar.STRING$.setField(ColumnType.scala:310) >> at >> org.apache.spark.sql.columnar.compression.RunLengthEncoding$Decoder.next(compressionSchemes.scala:168) >> at >> org.apache.spark.sql.columnar.compression.CompressibleColumnAccessor$class.extractSingle(CompressibleColumnAccessor.scala:37) >> at >> org.apache.spark.sql.columnar.NativeColumnAccessor.extractSingle(ColumnAccessor.scala:64) >> at >> org.apache.spark.sql.columnar.BasicColumnAccessor.extractTo(ColumnAccessor.scala:54) >> at >> org.apache.spark.sql.columnar.NativeColumnAccessor.org$apache$spark$sql$columnar$NullableColumnAccessor$$super$extractTo(ColumnAccessor.scala:64) >> at >> org.apache.spark.sql.columnar.NullableColumnAccessor$class.extractTo(NullableColumnAccessor.scala:52) >> at >> org.apache.spark.sql.columnar.NativeColumnAccessor.extractTo(ColumnAccessor.scala:64) >> at >> org.apache.spark.sql.columnar.InMemoryColumnarTableScan$$anonfun$9$$anonfun$14$$anon$2.next(InMemoryColumnarTableScan.scala:279) >> at >> org.apache.spark.sql.columnar.InMemoryColumnarTableScan$$anonfun$9$$anonfun$14$$anon$2.next(InMemoryColumnarTableScan.scala:275) >> at scala.collection.Iterator$$anon$13.next(Iterator.scala:372) >> at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) >> at scala.collection.Iterator$$anon$10.next(Iterator.scala:312) >> at scala.collection.Iterator$class.foreach(Iterator.scala:727) >> at scala.collection.AbstractIterator.foreach(Iterator.scala:1157) >> at >> scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48) >> at >> scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103) >> at >> scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47) >> at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273) >> at scala.collection.AbstractIterator.to(Iterator.scala:1157) >> at >> scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265) >> at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157) >> at >> scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252) >> at scala.collection.AbstractIterator.toArray(Iterator.scala:1157) >> at >> org.apache.spark.sql.execution.Limit$$anonfun$4.apply(basicOperators.scala:141) >> at >> org.apache.spark.sql.execution.Limit$$anonfun$4.apply(basicOperators.scala:141) >> at >> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314) >> at >> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314) >> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) >> at org.apache.spark.scheduler.Task.run(Task.scala:56) >> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) >> at >> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) >> at >> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) >> >> >> Any thoughts on this? Not sure if using the external metastore for the >> inital pull is a problem or if I'm just hitting a bug. >> >> >> >
