如何从DataFrame动态传递时间戳至Spark SQL查询并解决转换错误

错误原因 #

• 时间戳格式不兼容：你的时间戳用冒号分隔毫秒（2023-07-5 13:29:38:876），但SQL Server的日期时间类型要求毫秒部分用点分隔（2023-07-05 13:29:38.876），格式不匹配导致转换失败。
• 直接使用Row对象插值：ETLMetadatadate.head返回的是Row实例，直接拼到SQL里会生成不符合要求的字符串格式。
• SQL语法冗余：原查询中的外层括号和别名AS C属于多余内容，会引发语法错误。

修复步骤 #

1. 正确提取并格式化时间戳 #

从DataFrame中提取目标时间戳字段，转换为SQL Server兼容的格式：

// 假设metadatatable中存储时间戳的字段名为etl_timestamp
val ETLMetadatadate = spark.sql("select etl_timestamp from metadatatable limit 1")

// 如果原字段是Timestamp类型，直接格式化
val finaldateStr = ETLMetadatadate.selectExpr("date_format(etl_timestamp, 'yyyy-MM-dd HH:mm:ss.SSS')")
  .head()
  .getString(0)

// 如果原字段是字符串类型，先转Timestamp再格式化
// val finaldateStr = ETLMetadatadate.selectExpr("date_format(to_timestamp(etl_timestamp, 'yyyy-M-d HH:mm:ss:SSS'), 'yyyy-MM-dd HH:mm:ss.SSS')")
//   .head()
//   .getString(0)

2. 修正SQL查询语句 #

去掉冗余语法，确保时间戳格式正确：

val query = s"""SELECT
Ename,
ecompany,
sal
FROM Employee e
WHERE e.header_timestamp >= '$finaldateStr'"""

3. 修复Config配置语法 #

Map键值对之间需用逗号分隔：

val config = Config(Map(
  "url" -> "example.database.windows.net",
  "databaseName" -> "testdatabase",
  "querycustom" -> query,
  "user" -> "sample",
  "password" -> "1234",
  "connectionTimeout" -> "5",
  "querytimeout" -> "5"
))

4. 读取并写入数据 #

使用正确的Spark上下文执行操作：

import org.apache.spark.sql.SaveMode
val collection = spark.read.sqlDB(config)
collection.write.format("delta").mode(SaveMode.Overwrite).saveAsTable("testtable")

额外建议 #

• 优先使用参数化查询：避免直接字符串插值引发的SQL注入风险，若连接器支持参数传递，可通过参数绑定方式传入时间戳。
• 统一时间戳存储类型：源表中尽量将时间戳存储为Timestamp类型，而非字符串，减少格式转换问题。

内容的提问来源于stack exchange，提问作者bigdata techie