我是Spark和Scala的新手 . 我试图从SQL服务器中的过程中获取内容以在Spark SQL中使用它 . 为此,我通过Scala(Eclipse)中的JDBCRDD导入数据,并从该过程生成RDD .
创建RDD后,我将其注册为临时表,然后使用sqlContext.sql(“选择查询以选择特定列”) . 但是当我在select查询中输入列名时,它会抛出一个错误,因为我在RDD和临时表中都没有列名 .
请在下面找到我的代码:
val driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
val url = XXXX
val username = XXXX
val password = XXXX
val query = "select A, B, C, D from Random_Procedure where ID_1 = ? and ID_2 = ?"
// New SparkContext
val sc = new SparkConf().setMaster("local").setAppName("Amit")
val sparkContext = new SparkContext(sc)
val rddData = new JdbcRDD(sparkContext, () =>
DriverManager.getConnection(url, username, password),
query, 1, 0, 1, (x: ResultSet) => x.getString("A") + ", " +
x.getString("B") + ", " + x.getString("C") + ", " +
x.getString("D")).cache()
val sqlContext = new SQLContext(sparkContext)
import sqlContext.implicits._
val dataFrame = rddData.toDF
dataFrame.registerTempTable("Data")
sqlContext.sql("select A from Data").collect.foreach(println)
当我运行此代码时,它会抛出一个错误:无法解析给定输入列的“代码”_1;
但是当我运行:sqlContext.sql(“select * from Data”) . collect.foreach(println)它打印所有列A,B,C,D
我相信我没有在我创建的JdbcRDD中获取列名,因此它们在临时表中无法访问 . 我需要帮助 .
1 回答
问题是您创建了JdbcRDD对象,并且需要DataFrame . RDD simple不包含有关从元组到列名称的映射的信息 . 所以你应该从jdbc源创建DataFrame,如_1172410中所述 . 此外:
另请注意,Spark 1.3.0中添加了DataFrame . 如果您使用旧版本,则必须使用
org.apache.spark.sql.SchemaRDD