含LineString列的GeoDataFrame导入Oracle数据库遇类型不支持错误求助

前提条件 #

确保你的Oracle数据库已启用Oracle Spatial扩展，目标表的geometry列类型为SDO_GEOMETRY。

方法1：转WKT后通过Oracle函数构造空间对象 #

将Shapely的LineString转为WKT格式，插入时调用Oracle的SDO_GEOMETRY()函数转换为空间类型：

import pandas as pd
import sqlalchemy as sa
from sqlalchemy import types
from shapely.wkt import dumps

def push_dataframe(df_sql: pd.DataFrame, table_name, if_exists="replace", srid=4326):
    # 复制数据避免修改原GeoDataFrame
    df = df_sql.copy()
    # 将LineString转为WKT字符串
    df['geometry'] = df['geometry'].apply(lambda geom: dumps(geom) if geom is not None else None)
    
    dict_dtype = {}
    for col in df.columns:
        if pd.api.types.is_string_dtype(df[col]):
            dict_dtype[col] = types.VARCHAR(255)
    # 为WKT字符串指定足够长度的字段类型
    dict_dtype['geometry'] = types.VARCHAR(4000)  # WKT过长时可改用CLOB类型

    oracle_db = sa.create_engine(
        "oracle://USER:PW@ZZ.YY.local/XX"
    )

    with oracle_db.connect() as connection:
        # 替换模式下手动创建表，确保geometry列类型正确
        if if_exists == "replace":
            connection.execute(sa.text(f"DROP TABLE IF EXISTS \"SCHEMA\".{table_name.lower()}"))
            create_table_sql = f"""
            CREATE TABLE \"SCHEMA\".{table_name.lower()} (
                x1 VARCHAR2(255),
                x2 VARCHAR2(255),
                geometry SDO_GEOMETRY
            )
            """
            connection.execute(sa.text(create_table_sql))
        
        # 批量插入时用SDO_GEOMETRY转换WKT
        insert_sql = sa.text(f"""
            INSERT INTO \"SCHEMA\".{table_name.lower()}(x1, x2, geometry)
            VALUES (:x1, :x2, SDO_GEOMETRY(:geometry, {srid}))
        """)
        chunksize = 1000
        for i in range(0, len(df), chunksize):
            chunk = df.iloc[i:i+chunksize]
            params = chunk.to_dict('records')
            connection.execute(insert_sql, params)
        connection.commit()
    return

方法2：自定义SQLAlchemy类型处理SDO_GEOMETRY #

定义自定义类型自动完成Shapely对象与Oracle空间类型的双向转换：

import sqlalchemy as sa
from sqlalchemy.types import UserDefinedType
from shapely.wkt import loads, dumps
import cx_Oracle

class SDOGeometry(UserDefinedType):
    def get_col_spec(self):
        return "SDO_GEOMETRY"
    
    def bind_processor(self, dialect):
        def process(value):
            if value is None:
                return None
            # 直接构造Oracle SDO_GEOMETRY对象
            return cx_Oracle.Object(
                "SDO_GEOMETRY", dialect.dbapi_connection,
                {
                    "SDO_GTYPE": 2002,  # 2维LineString的GTYPE，3维用3002
                    "SDO_SRID": 4326,
                    "SDO_POINT": None,
                    "SDO_ELEM_INFO": cx_Oracle.Array([1,2,1], dialect.dbapi_connection),
                    "SDO_ORDINATES": cx_Oracle.Array(
                        [coord for pt in value.coords for coord in pt], 
                        dialect.dbapi_connection
                    )
                }
            )
        return process
    
    def result_processor(self, dialect, coltype):
        def process(value):
            if value is None:
                return None
            # 将SDO_GEOMETRY转回Shapely对象（读取数据时用）
            wkt = f"LINESTRING({','.join([f'{x} {y}' for x,y in zip(value.SDO_ORDINATES[::2], value.SDO_ORDINATES[1::2])])})"
            return loads(wkt)
        return process

# 修改后的推送函数
def push_dataframe(df_sql: pd.DataFrame, table_name, if_exists="replace"):
    dict_dtype = {}
    for col in df_sql.columns:
        if pd.api.types.is_string_dtype(df_sql[col]):
            dict_dtype[col] = types.VARCHAR(255)
    # 指定geometry列使用自定义类型
    dict_dtype['geometry'] = SDOGeometry()
    
    oracle_db = sa.create_engine(
        "oracle://USER:PW@ZZ.YY.local/XX"
    )
    
    with oracle_db.connect() as connection:
        df_sql.to_sql(
            table_name.lower(),
            connection,
            schema="SCHEMA",
            if_exists=if_exists,
            index=False,
            chunksize=1000,
            dtype=dict_dtype,
        )
    return

关键注意事项 #

• GTYPE调整：2002对应2维LineString，若使用3维数据需改为3002，其他几何类型需对应调整GTYPE值。
• SRID匹配：替换代码中的4326为你实际使用的空间参考ID（如CGCS2000对应4490）。
• 长WKT处理：若LineString的WKT长度超过4000字符，需将字段类型改为CLOB避免数据截断。

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