我正在尝试使用 Spark 读取 greenplum 数据库上存在的表,如下所示:
val execQuery = s"select ${allColumns}, 0 as ${flagCol} from schema.table where period_year=2017 and period_num=12"
val yearDF = spark.read.format("io.pivotal.greenplum.spark.GreenplumRelationProvider").option("url", connectionUrl).option("dbtable", s"(${execQuery}) as year2016")
.option("user", devUserName)
.option("password", devPassword)
.option("partitionColumn","header_id")
.option("lowerBound", 16550)
.option("upperBound", 1152921481695656862L)
.option("numPartitions",450).load()
当我使用 Spark-submit 运行代码时,出现异常:
Exception in thread "main" org.postgresql.util.PSQLException: ERROR: relation "public.(select je_header_id,source_system_name,je_line_num,last_update" does not exist
Position: 15
at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2310)
at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2023)
at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:217)
at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:421)
at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:318)
at org.postgresql.jdbc.PgStatement.executeQuery(PgStatement.java:281)
at com.zaxxer.hikari.pool.ProxyStatement.executeQuery(ProxyStatement.java:111)
at com.zaxxer.hikari.pool.HikariProxyStatement.executeQuery(HikariProxyStatement.java)
at io.pivotal.greenplum.spark.jdbc.Jdbc$.resolveTable(Jdbc.scala:301)
at io.pivotal.greenplum.spark.GreenplumRelationProvider.createRelation(GreenplumRelationProvider.scala:29)
at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:309)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:178)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:146)
at com.partition.source.YearPartition$.prepareFinalDF$1(YearPartition.scala:141)
at com.partition.source.YearPartition$.main(YearPartition.scala:164)
at com.partition.source.YearPartition.main(YearPartition.scala)
In the execQuery
我可以看到模式名称和表名称正确形成。当我提交代码时,它说public.(select je_header_id,source_system_name,) relation not found
。我不明白为什么要采取public
作为模式名称和查询(select je_header_id,source_system_name,je_line_num,last_update"
作为表名。
谁能让我知道我在这里犯了什么错误以及如何解决它?