我在spark-cassandra-connector中使用Spark 1.2.1:
//join with cassandra
val rdd = some_array.map(x => SomeClass(x._1,x._2)).joinWithCassandraTable(keyspace, some_table)
println(timer, "Join")
//get only the jsons and create rdd temp table
val jsons = rdd.map(_._2.getString("this"))
val jsonSchemaRDD = sqlContext.jsonRDD(jsons)
jsonSchemaRDD.registerTempTable("this_json")
println(timer, "Map")
输出为:
Timer "Join"- 558 ms
Timer "Map"- 290284 ms
我猜想“ joinWithCassandraTable()”函数是惰性的,如果是这样,那是什么原因呢?
最佳答案
实际上,将在此处触发评估的部分是sqlContext.jsonRDD
。由于不提供schema
,因此必须实现jsons
才能进行推断。joinWithCassandraTable
有点类似,因为它必须连接到Cassandra并获取所需的元数据。见Apache Spark: Driver (instead of just the Executors) tries to connect to Cassandra