我正在尝试将apachephoenix的数据加载到sparkDataframe中。
我已经成功地用以下代码创建了rdd:
val sc = new SparkContext("local", "phoenix-test")
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val foo: RDD[Map[String, AnyRef]] = sc.phoenixTableAsRDD(
table = "FOO",
columns = Seq("ID", "MESSAGE_EPOCH", "MESSAGE_VALUE"),
zkUrl = Some("<zk-ip-address>:2181:/hbase-unsecure"))
foo.collect().foreach(x => println(x))
然而,我并没有那么幸运地尝试创建一个Dataframe。我目前的尝试是:
val sc = new SparkContext("local", "phoenix-test")
val sqlContext = new SQLContext(sc)
val df = sqlContext.phoenixTableAsDataFrame(
table = "FOO",
columns = Seq("ID", "MESSAGE_EPOCH", "MESSAGE_VALUE"),
zkUrl = Some("<zk-ip-address>:2181:/hbase-unsecure"))
df.select(df("ID")).show
不幸的是,上面的代码导致classcastexception:
java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericMutableRow cannot be cast to org.apache.spark.sql.Row
我还是个新手。如果有人能帮忙,我们将不胜感激!
1条答案
按热度按时间hfyxw5xn1#
虽然你没有提到你的Spark版本和例外的细节。。。
请看Phoenix-2287,它是固定的,上面写着
环境:在os x上以独立模式运行的hbase 1.1.1*
spark 1.5.0Phoenix城4.5.2
josh mahonin添加了一条评论-23/sep/15 17:56更新的补丁增加了对spark 1.5.0的支持,并且向后兼容到1.3.0(手动测试,spark版本概要文件可能值得在将来查看),在1.5.0中,它们已经消失并显式隐藏了
GenericMutableRow
数据结构。幸运的是,我们能够Row
'数据类型,它是向后兼容的,并且在将来的版本中也应该保持兼容。作为更新的一部分,sparksql不推荐在其DecimalType
'. 在更新这个时,我暴露了一个新的问题,那就是我们没有将基础decimal类型的精度和规模推进到spark。目前,我已经将其设置为使用spark默认值,但我将专门为此创建另一个问题。我在这个补丁中也包含了一个被忽略的集成测试。