Received an invalid column length from the bcp client in spark job

I am playing around with spark and wanted to store a data frame in a sql database. It works but not when saving a datetime column:

from pyspark.sql import SparkSession,Row
from pyspark.sql.types import IntegerType,TimestampType,StructType,StructField,StringType
from datetime import datetime

...

spark = SparkSession.builder \
    ...
    .getOrCreate()
    
# Create DataFrame 

rdd = spark.sparkContext.parallelize([
    Row(id=1, title='string1', created_at=datetime.now())
])

schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("title", StringType(), False),
    StructField("created_at", TimestampType(), True)
])

df = spark.createDataFrame(rdd, schema)

df.show()

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("overwrite") \
    .option("truncate", True) \
    .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Schema:

Error:

com.microsoft.sqlserver.jdbc.SQLServerException: Received an invalid column length from the bcp client for colid 3

From my understanding the error states that datetime.now() has invalid length. But how can that be, if it is a standard datetime? Any ideas what the issue is?

SQLServer datetime datatype has time range - 00:00:00 through 23:59:59.997. output of datetime.now() will not fit in for datetime, you need to change the datatype on SQLSever table to datetime2

There are problems with the code to create the dataframe. You are missing libraries. The code below creates the dataframe correctly.

#
#  1 - Make test dataframe
#

# libraries
from pyspark.sql import Row
from datetime import datetime
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

# create rdd
rdd = spark.sparkContext.parallelize([Row(id=1, title='string1', created_at=datetime.now())])

# define structure
schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("title", StringType(), False),
    StructField("created_at", TimestampType(), True)
])

# create df
df = spark.createDataFrame(rdd, schema)

# show df
display(df)

The output is shown above. We need to create a table that follows the nullability and data types.

The code below creates a table called stack_overflow.

-- drop table
drop table if exists stack_overflow
go

-- create table
create table stack_overflow
(
  id int not null,
  title varchar(100) not null,
  created_at datetime2 null
)
go

-- show data
select * from stack_overflow
go

Next, we need to define our connection properties.

#
#  2 - Set connection properties
#

server_name = "jdbc:sqlserver://svr4tips2030.database.windows.net"
database_name = "dbs4tips2020"
url = server_name + ";" + "databaseName=" + database_name + ";"
user_name = "jminer"
password = "<your password here>"
table_name = "stack_overflow"

Last, we want to execute the code to write the dataframe.

#
#  3 - Write test dataframe
#

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("overwrite") \
    .option("truncate", True) \
    .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", user_name) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Executing a select query shows that the data was written correctly.

In short, look at the documentation for Spark SQL Types. I found out that datetime2 works nicely.

https://learn.microsoft.com/en-us/dotnet/api/microsoft.spark.sql.types?view=spark-dotnet

One word of caution, this code does not handle date time offset. Also, there is no data type in Spark to use in an offset mapping.

# Sample date time offset value
import pytz
from datetime import datetime, timezone, timedelta
user_timezone_setting = 'US/Pacific'
user_timezone = pytz.timezone(user_timezone_setting)
the_event = datetime.now() 
localized_event = user_timezone.localize(the_event)
print(localized_event)

The code above creates a variable with the following data.

But once we cast it to a dataframe, it loses the offset since it is converted to UTC time. If UTC offset is important, you will have to pass that information as separate integer.