使用带有 jdbc 写入的 PySpark 数据框在 PostgreSQL 上编写修女时出现问题

Question

所以我正在使用 PySpark（Spark 3.0.1、Scala 2.12）将数据从 MySQL (5.7) 数据库移动到 PostgreSQL (12.7) 数据库。命运模型中的表有一个 Enum 列。

CREATE TYPE ORDER_STATUS AS ENUM (
'SHIPPED','PAID','REFUNDED','PARTIALLY_REFUNDED','PROCESSING');

插入时：

df_orders.select(df_orders.columns).write.format('jdbc').options(**postgres_write_opts_table).mode('append').save()

我得到下一个异常

Caused by: org.postgresql.util.PSQLException: ERROR: column "status" is of type order_status but expression is of type character varying
Hint: You will need to rewrite or cast the expression.

基本上我需要将列状态转换为 ORDER_STATUS 。我曾尝试使用UserDefinedType（PySpark 没有SQLUserDefinedType），但我不知道我在做什么，因为文档不是很清楚。

class StatusUDT(UserDefinedType):
@classmethod
def sqlType(self):
    return NullType()

@classmethod
def module(cls):
    return cls.__module__

def serialize(self, obj):
    return f"{obj.value}::order_status_type"

def deserialize(self, datum):
    return {x.value: x for x in Some}[datum]

然后我尝试铸造

df_orders = df_orders.withColumn("status", col("status").cast(StatusUDT()))

然后我收到下一个错误：

AnalysisException: cannot resolve 'CAST(`status` AS NULL)' due to data type mismatch: cannot cast string to null;;

有什么方法可以转换这个 Enum 吗？

Answer 1

所以我终于能够克服这个问题。我暂时删除了 Enum，这样我就可以继续做更多的测试，然后我遇到了 JSON 类型的类似问题。 搜索它我发现了这个帖子：How to save String as JSONB type in postgres when using AWS Glue。我修复了它设置属性：

'stringtype':"unspecified"

正如帖子答案所暗示的那样。

然后我将 Enum 放回表中，这个属性也起作用了。我能够毫无问题地运行插入。