SparkSQL和UDT

Question

我尝试使用SparkSQL (v.1.3.0)访问PostgreSQL数据库。在这个数据库中，我有一张表

CREATE TABLE test (
 id bigint,
 values double precision[]
);

对于访问表，我使用

val sparkConf = new SparkConf().setAppName("TestRead").setMaster("local[2]")
val sc = new SparkContext(sparkConf)
val sqlContext = new SQLContext(sc)

val jdbcDF = sqlContext.load("jdbc", Map(
  "url" -> "jdbc:postgresql://...",
  "dbtable" -> "schema.test",
  "user" -> "...",
  "password" -> "..."))

sqlContext.sql("SELECT * FROM schema.test")

但是，每次我尝试访问包含这个数组的表时，我都会得到一个java.sql.SQLException: Unsupported type 2003。

我在星火测试代码中找到了一个例子，它在星火中为一个二维点创建了一个UDT (参见ExamplePointUDT.scala)。但是，我不明白我怎么可能使用这段代码。

Answer 1

这是可以实现的，至少可以通过在查询中进行强制转换来实现，至少在pyspark中是如此。不要让不受支持的类型达到火花，将它们抛到数据库中，然后在得到表后再将它们抛回。

我不确定语法是否正确，但应该是这样的：

val query_table = "(SELECT id, CAST(values AS TEXT) FROM schema.test) AS casted_table"

val jdbcDF = sqlContext.load("jdbc", Map(
  "url" -> "jdbc:postgresql://...",
  "dbtable" -> query_table,
  "user" -> "...",
  "password" -> "..."))

jdbcDF.map(x => (x.id, x.values.toArray))

我很确定没有.toArray可以将字符串表示转换回数组，它只是位置保持代码。但现在这只是正确解析它的问题。

当然，这只是一个补丁，但它有效。