我正在尝试使用具有指定偏移量的 Spark jdbc 从 MSSQL 数据库中读取数据。所以数据应该只在指定的时间戳之后加载,这就是这个偏移量。我试图通过在 jdbc 配置中提供查询来实现它,但是,我没有发现用参数化值创建准备好的语句的可能性。在这种情况下,我想参数化一个偏移量,该偏移量在每次应用程序启动后都会改变。如何使用 jdbc 选项实现它?
所有数据库配置都驻留在 application.conf 文件中。这是我从数据库中读取的方式:
def jdbcOptions(query: String) = Map[String,String](
"driver" -> config.getString("sqlserver.db.driver"),
"url" -> config.getString("sqlserver.db.url"),
"dbtable" -> s"(select * from TestAllData where update_database_time >= '2019-03-19 12:30:00.003') as subq,
"user" -> config.getString("sqlserver.db.user"),
"password" -> config.getString("sqlserver.db.password"),
"customSchema" -> config.getString("sqlserver.db.custom_schema")
)
val testDataDF = sparkSession
.read
.format("jdbc")
.options(jdbcOptions())
.load()
相反,查询应该看起来像这样:
s"(select * from TestAllData where update_database_time >= $tmstp) as subq