AWS Glue中filter方法引发"JDBC type for null"错误求助

错误信息 #

写入SQL Server RDS时触发以下错误：

Can't get JDBC type for null Failed

代码示例 #

def get_credentials(self, glue_connection):
    glue_context = self.gc
    credentials = glue_context.extract_jdbc_conf(glue_connection)

    url = credentials.get("url")

    host = url[url.find("//") + 2 : url.rfind(":")]
    port = url[url.rfind(":") + 1 :]

    normalized_credentials = {
        "url": credentials.get("url"),
        "host": host,
        "port": port,
        "user": credentials.get("user"),
        "password": credentials.get("password"),
    }

    return normalized_credentials

def create_source_df(self, schema_table):

    credentials = self.get_credentials(self.landing_connection_name)

    mssql_df = self.gc.create_dynamic_frame.from_options(
        connection_type="sqlserver",
        connection_options={
            "url": credentials["url"] + ";database=" + self.database_name,
            "user": credentials["user"],
            "password": credentials["password"],
            "dbtable": schema_table,
        },
        transformation_ctx="glue_df",
    )

    return mssql_df

glue_df = glue_obj.create_source_df(schema_table)
glue_df = glue_df.resolveChoice(specs=[("created_at", "cast:date")])
glue_df = glue_df.resolveChoice(specs=[("updated_at", "cast:date")])

# 执行此过滤后写入RDS触发错误
glue_df = glue_df.filter(f=lambda x: x["filed_to_filter"] not in ["DELETE"])

解决方案 #

原因分析 #

过滤操作后，可能出现以下情况导致报错：

• 部分字段在过滤后所有值均为null，JDBC驱动无法推断该字段对应的SQL数据类型
• Glue DynamicFrame在执行过滤后丢失了字段的元数据类型信息

具体解决办法 #

• 显式保留字段类型：过滤后重新对可能出现全null的字段执行类型转换，确保元数据不丢失

  # 过滤后重新指定filed_to_filter字段类型为字符串
  glue_df = glue_df.filter(f=lambda x: x["filed_to_filter"] not in ["DELETE"])
  glue_df = glue_df.resolveChoice(specs=[("filed_to_filter", "cast:string")])
  

• 改用Spark DataFrame执行过滤：Glue DynamicFrame的filter可能存在元数据丢失问题，转为Spark DataFrame操作后再转回DynamicFrame

  from awsglue.dynamicframe import DynamicFrame
  
  # 转为Spark DataFrame
  spark_df = glue_df.toDF()
  # 执行过滤
  spark_df = spark_df.filter(spark_df["filed_to_filter"] != "DELETE")
  # 转回Glue DynamicFrame
  glue_df = DynamicFrame.fromDF(spark_df, self.gc, "filtered_dynamic_frame")
  

• 提前处理null值：在过滤前对目标字段的null值进行填充，避免过滤后出现全null字段

  # 给filed_to_filter字段的null值填充默认值
  glue_df = glue_df.fillna({"filed_to_filter": "UNKNOWN"})
  # 再执行过滤
  glue_df = glue_df.filter(f=lambda x: x["filed_to_filter"] not in ["DELETE"])
  

• 验证目标表结构：确认SQL Server目标表中对应字段的类型允许null，且与Glue DataFrame的字段类型完全匹配

• 检查过滤后的Schema：执行glue_df.printSchema()查看过滤后的字段类型，确保所有字段都保留了正确的类型信息

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