如何从PySpark DataFrame生成GeoJSON？规避Pandas/GeoPandas低效问题

问题根源 #

你之前的SQL实现出现格式错误，核心原因是重复进行了JSON序列化：

• ST_AsGeoJSON已经将几何对象转为JSON格式的字符串
• 外层又用to_json包裹，导致这个JSON字符串被再次转义成普通字符串（出现多层\转义）
• 同理properties部分的to_json也重复序列化，最终生成的Feature里的geometry和properties都是字符串，而非合法的JSON对象

正确实现方案 #

方案1：基于Spark原生结构体构建（推荐） #

直接用Spark原生类型构建Feature结构，几何部分通过from_json解析ST_AsGeoJSON的输出为Spark结构体，避免重复序列化：

# 构建单个Feature的SQL查询
data_union_json_query = f"""
SELECT 
    to_json(named_struct(
        'type', 'Feature',
        'geometry', from_json(
            ST_AsGeoJSON(ST_Transform(ST_SetSRID(ST_GeomFromText(geom), 27700), 'EPSG:4326')),
            -- 定义GeoJSON几何的Schema，适配MultiPolygon类型，其他类型可调整
            struct(
                'type' string,
                'coordinates' array<array<array<array<double>>>>
            )
        ),
        'properties', named_struct(
            'id', id,
            'initial', initial,
            'partitiongroup', partitiongroup
        )
    )) AS feature
FROM 
    partition_to_process_view
WHERE 
    partitiongroup = '{partition}'
"""

# 执行SQL获取每个Feature的JSON字符串
data_json_df = spark.sql(data_union_json_query)

# 构建标准FeatureCollection结构
feature_collection_df = data_json_df.agg(
    to_json(named_struct(
        'type', 'FeatureCollection',
        'features', collect_list(from_json(col('feature'), struct(
            'type' string,
            'geometry' struct('type' string, 'coordinates' array<array<array<array<double>>>>),
            'properties' struct('id' string, 'initial' string, 'partitiongroup' integer)
        )))
    )).alias('geojson')
)

# 导出为单个GeoJSON文件
feature_collection_df.coalesce(1).write.mode('overwrite').text(f's3://my-bucket/geojson-folder/myname-{folder}.geojson')

方案2：字符串拼接快速实现 #

如果不想定义复杂Schema，可直接通过字符串拼接构建Feature，避免重复序列化问题：

# 构建单个Feature的SQL查询
data_union_json_query = f"""
SELECT 
    concat(
        '{{"type": "Feature",',
        '"geometry": ', ST_AsGeoJSON(ST_Transform(ST_SetSRID(ST_GeomFromText(geom), 27700), 'EPSG:4326')), ',',
        '"properties": ', to_json(named_struct('id', id, 'initial', initial, 'partitiongroup', partitiongroup)),
        '}}'
    ) AS feature
FROM 
    partition_to_process_view
WHERE 
    partitiongroup = '{partition}'
"""

# 执行SQL获取每个Feature的JSON字符串
data_json_df = spark.sql(data_union_json_query)

# 拼接成标准FeatureCollection
feature_collection_df = data_json_df.agg(
    concat(
        '{{"type": "FeatureCollection", "features": [',
        concat_ws(',', collect_list(col('feature'))),
        ']}'
    ).alias('geojson')
)

# 导出为单个GeoJSON文件
feature_collection_df.coalesce(1).write.mode('overwrite').text(f's3://my-bucket/geojson-folder/myname-{folder}.geojson')

关键注意事项 #

• 禁止嵌套使用to_json：ST_AsGeoJSON输出的是合法JSON字符串，直接作为geometry的值即可
• 必须生成FeatureCollection结构：单独的Feature行不符合GeoJSON标准，多数GIS工具无法识别
• coalesce(1)仅用于生成单个文件，大数据量时谨慎使用（可按分区生成多个GeoJSON文件）
• 几何Schema需匹配：如果存在多种几何类型（Point、Polygon等），可改用array<object>等通用类型适配

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