NetSuite Suite Analytics JDBC timestamp字段致Databricks空指针异常

在Databricks集群通过Suite Analytics JDBC连接NetSuite，查询classification表的lastmodifieddate（timestamp类型）字段时触发NullPointerException，但在DBeaver中查询该表未发现空值行。尝试用COALESCE函数处理该字段时，又出现SQL执行错误。

相关代码 #

jdbc_url = "jdbc:ns://x.connect.api.netsuite.com:1708;ServerDataSource=NetSuite2.com;Encrypted=1;NegotiateSSLClose=false;CustomProperties=(AccountID=x;RoleID=1030)"
driver = "com.netsuite.jdbc.openaccess.OpenAccessDriver"

secret_scope = "fl-da-kv-app-scope"
secret_key_password = "netsuite-password"

user =  "user"
password = dbutils.secrets.get(scope=secret_scope, key=secret_key_password)

netsuite_query = """
SELECT
lastmodifieddate
# CAST(lastmodifieddate AS VARCHAR(255)) AS lastmodifieddate_str
# COALESCE(lastmodifieddate, TO_DATE('1900-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS')) AS lastmodifieddate
FROM "Fl - Accountant".classification
"""


# netsuite_query = """
# SELECT
#     custrecord_nspbcs_class_planning_cat,
#     custrecord_lmry_class_code,
#     externalid,
#     fullname,
#     id,
#     includechildren,
#     isinactive,
#     name,
#     parent,
#     subsidiary
# FROM "Fl - Accountant".classification
# """

# netsuite_query = """SELECT custrecord_lmry_class_code FROM "Fl- Accountant".classification"""
# netsuite_query = """
# SELECT
# COALESCE(custrecord_lmry_class_code, 'a') AS custrecord_lmry_class_code,
# COALESCE(custrecord_nspbcs_class_planning_cat, 0) AS custrecord_nspbcs_class_planning_cat,

# id as id

# FROM "Fl - Accountant".classification
# """

output_path_csv = "/tmp/netsuite_classification_full1.csv"

# output_path_delta = "/tmp/delta/netsuite_classification"


try:
    print("Attempting to read full table from NetSuite...")
    print(f"Executing query: {netsuite_query}")


    df = spark.read \
        .format("jdbc") \
        .option("url", jdbc_url) \
        .option("query", netsuite_query) \
        .option("user", user) \
        .option("password", password) \
        .option("driver", driver) \
        .load()

    print("\nSUCCESS: Read data from NetSuite into a DataFrame.")

    # if theres an error w. the count that means the query is wrong.
    row_count = df.count()
    print(f"SUCCESS: Read {row_count} rows from NetSuite... ")
    # df.show(5)

    print(f"\nAttempting to write data to CSV at: {output_path_csv}")
    # df_cleaned = df.na.fill('')
    # print('cleaned the df')

    df.write \
      .format("csv") \
      .option("header", "true") \
      .mode("overwrite") \
      .save(output_path_csv)

    print(f"SUCCESS: Wrote data to CSV.")

    # # --- Write to Delta Table ---
    # print(f"\nAttempting to write data to Delta table at: {output_path_delta}")
    # df.write \
    #   .format("delta") \
    #   .mode("overwrite") \
    #   .save(output_path_delta)
    # print(f"SUCCESS: Wrote data to Delta table.")


except Exception as e:
    print("\nERROR: An exception occurred.")
    # The full error will be raised, giving you the complete stack trace.
    raise e

使用COALESCE的查询语句 #

SELECT
COALESCE(lastmodifieddate, TO_DATE('1900-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS')) AS lastmodifieddate
FROM "Fl - Accountant".classification

报错信息 #

尝试从NetSuite读取全表数据... 执行查询：
SELECT COALESCE(lastmodifieddate, TO_DATE('1900-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS')) AS lastmodifieddate FROM "Fl - Accountant".classification

任务因阶段失败而中止：阶段163.0中的任务0失败4次，最近一次失败：丢失阶段163.0中的任务0.3（TID 222）（10.21.40.204执行器0）：java.sql.SQLException: [NetSuite][OpenAccess SDK JDBC Driver][OpenAccess SDK SQL Engine]无法检索数据。错误工单# mcwb7tcx1eia537oa41ma[400]
at com.netsuite.jdbc.base.dt.b(oajc:1099)
at com.netsuite.jdbc.base.dt.a(oajc:976)
at com.netsuite.jdbc.base.dt.a(oajc:1120)
at com.netsuite.jdbc.base.dt.a(oajc:187)
at com.netsuite.openaccess.ssp.at.s(Unknown Source)
at com.netsuite.openaccess.ssp.f.a(Unknown Source)
at com.netsuite.openaccess.ssp.f.R(Unknown Source)
at com.netsuite.openaccess.ssp.f.rS(Unknown Source)
at com.netsuite.openaccess.ssp.f.rR(Unknown Source)
at com.netsuite.openaccess.ssp.f.E(Unknown Source)
at com.netsuite.openaccess.ctxt.stmt.c.a(Unknown Source)
at com.netsuite.jdbc.openaccess.d.eR(Unknown Source)
at com.netsuite.jdbc.base.hd.kW(oajc:2510)
at com.netsuite.jdbc.base.hd.kV(oajc:2397)
at com.netsuite.jdbc.base.fp.executeQuery(oajc:479)
at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD.compute(JDBCRDD.scala:404)

at org.apache.spark.sql.execution.datasources.FileFormatWriter$.executeWrite(FileFormatWriter.scala:431)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.$anonfun$write$1(FileFormatWriter.scala:300)
at com.databricks.spark.util.FrameProfiler$.record(FrameProfiler.scala:94)

at org.apache.spark.scheduler.Task.$anonfun$run$5(Task.scala:161)
at com.databricks.unity.UCSEphemeralState$Handle.runWith(UCSEphemeralState.scala:51)
at com.databricks.unity.HandleImpl.runWith(UCSHandle.scala:104)
at com.databricks.unity.HandleImpl.$anonfun$runWithAndClose$1(UCSHandle.scala:109)
at scala.util.Using$.resource(Using.scala:269)
at com.databricks.unity.HandleImpl.runWithAndClose(UCSHandle.scala:108)
at org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:155)
at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.scheduler.Task.run(Task.scala:102)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$10(Executor.scala:1038)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:112)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:1041)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:928)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)

文件 , 行92
90 print("\nERROR: An exception occurred.")
91 # 抛出完整错误栈以排查问题
---> 92 raise e
文件 , 行76
68 print(f"\nAttempting to write data to CSV at: {output_path_csv}")
69 # df_cleaned = df.na.fill('')
70 # print('cleaned the df')
72 df.write

73 .format("csv")

74 .option("header", "true")

75 .mode("overwrite")

---> 76 .save(output_path_csv)
78 print(f"SUCCESS: Wrote data to CSV.")
80 # # --- 写入Delta表 ---
81 # print(f"\nAttempting to write data to Delta table at: {output_path_delta}")
82 # df.write \ (...)
85 # .save(output_path_delta)
86 # print(f"SUCCESS: Wrote data to Delta table.")

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