我试图从sqlserver读取一个表,并在读取时应用分区。在读取数据之前,我想得到lowerBound&上界如下。
boundsDF = spark.read.format('jdbc')
.option('url', 'url')
.option('driver', 'com.microsoft.sqlserver.jdbc.SQLServerDriver')
.option('user', username)
.option('password', password)
.option('dbtable', f'(select min(updated_datetime) as mint, max(updated_datetime) as maxt from tablename)
.load()
我从boundsDF中提取了如下值:
maxdate = [x["maxt"] for x in boundsDF.rdd.collect()]
mindate = [x["mint"] for x in boundsDF.rdd.collect()]
这就是我在阅读时指定时间戳列的方式:
dataframe = spark.read.format('jdbc')
.option('url', url)
.option('driver', 'com.microsoft.sqlserver.jdbc.SQLServerDriver')
.option('user', user)
.option('password', password)
.option('dbtable', tablename)
.option('partitionColumn', timestamp_column)
.option('numPartitions', 3)
.option('lowerBound', mindate[0])
.option('upperBound', maxdate[0])
.option('fetchsize', 5000)
.load()
如果我打印mindate&maxdate如下所示:
mindate[0]: datetime.datetime(2010, 10, 4, 11, 54, 13, 543000)
maxdate[0]: datetime.datetime(2021, 3, 5, 17, 59, 45, 880000)
当我打印的时候
dataframe.count()
,我看到一条异常消息,如下所示。
例外情况:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 18.0 failed 1 times, most recent failure: Lost task 2.0 in stage 18.0 (TID 21, executor driver): com.microsoft.sqlserver.jdbc.SQLServerException: Conversion failed when converting date and/or time from character string.
自从我开始使用Spark以来,我一直使用整数列作为分区列。这是我第一次使用时间戳列对数据进行分区。
是mindate[0]&maxdate[0]的格式是否在我的read语句中指定?
有人能告诉我我是否以正确的方式实现了代码吗?
