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.

Я нашел пример в тестовом коде Spark, который создает UDT в Spark для двумерной точки (см. 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, что будет отбрасывать строковое представление обратно в массив, это просто введите код владельца. Но теперь это просто вопрос правильного его разбора.

Конечно, это всего лишь патч, но он работает.