DynamoDB多边形存储与空间查询相关技术问询

1. 多边形等复杂几何图形的存储方案 #

官方DynamoDB Geo Library仅支持点数据，存储多边形需要自定义表结构，核心思路是结合Geohash分区与几何元数据存储：

• 表结构设计：
  • PK：用多边形bounding box对应的Geohash前缀（比如6-8位精度）作为分区键，确保空间邻近的多边形落在同一/相邻分区，减少后续查询范围。
  • SK：用多边形唯一ID（如POLYGON#UUID）作为排序键，方便单多边形的快速查询。
  • 属性字段：
    • geo_json：存储多边形的GeoJSON字符串（直接序列化几何对象，方便后续解析）。
    • bbox：存储bounding box的四元组（min_lon, min_lat, max_lon, max_lat），类型为Number数组，用于初步过滤。
• 存储逻辑：
  先计算多边形的bounding box，生成覆盖该bbox的所有Geohash前缀（避免单个大多边形跨多个分区时漏查），将多边形数据写入每个对应Geohash前缀的分区中。

2. 复杂空间查询（如相交统计）的实现思路 #

DynamoDB无原生空间索引，需通过"粗过滤+精匹配"两步实现：

• 粗过滤：计算目标几何图形的bounding box，生成覆盖该bbox的所有Geohash前缀，批量查询这些前缀对应的DynamoDB分区，得到候选多边形集合（仅筛选bounding box有重叠的对象，减少后续计算量）。
• 精匹配：在应用层使用Python空间库（如Shapely）解析候选多边形的GeoJSON，调用空间关系判断方法（如intersects()），统计与目标几何图形相交的多边形数量。
• 优化点：如果数据量极大，可结合DynamoDB的FilterExpression先过滤掉bbox完全不重叠的项，但注意FilterExpression是扫描后过滤，效率不如Geohash分区前置过滤。

3. Python实现的具体步骤 #

由于官方无Python版Geo Library，需自行实现Geohash处理与空间逻辑：

1. 依赖库安装：

   pip install geohash shapely boto3
   

2. 表结构创建（用boto3）：

   import boto3
   
   dynamodb = boto3.resource('dynamodb')
   table = dynamodb.create_table(
       TableName='GeoPolygons',
       KeySchema=[
           {'AttributeName': 'PK', 'KeyType': 'HASH'},
           {'AttributeName': 'SK', 'KeyType': 'RANGE'}
       ],
       AttributeDefinitions=[
           {'AttributeName': 'PK', 'AttributeType': 'S'},
           {'AttributeName': 'SK', 'AttributeType': 'S'}
       ],
       ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
   )
   table.wait_until_exists()
   

3. 多边形存储逻辑：

   import geohash
   from shapely.geometry import shape
   import json
   
   def store_polygon(polygon_geojson, polygon_id):
       geom = shape(polygon_geojson)
       bbox = geom.bounds  # (min_lon, min_lat, max_lon, max_lat)
       # 生成覆盖bbox的Geohash前缀（精度6位）
       geohash_prefixes = get_geohash_prefixes(bbox, precision=6)
       for prefix in geohash_prefixes:
           table.put_item(
               Item={
                   'PK': prefix,
                   'SK': f'POLYGON#{polygon_id}',
                   'geo_json': json.dumps(polygon_geojson),
                   'bbox': list(bbox)
               }
           )
   
   def get_geohash_prefixes(bbox, precision):
       # 计算bbox四角的Geohash，去重后得到所有覆盖的前缀
       min_lon, min_lat, max_lon, max_lat = bbox
       corners = [
           (min_lat, min_lon), (min_lat, max_lon),
           (max_lat, min_lon), (max_lat, max_lon)
       ]
       prefixes = set()
       for lat, lon in corners:
           gh = geohash.encode(lat, lon, precision=precision)
           prefixes.add(gh)
       # 补充边缘覆盖的前缀（可选，避免bbox跨前缀时漏查）
       return list(prefixes)
   

4. 相交查询与统计：

   def count_intersecting_polygons(target_geojson):
       target_geom = shape(target_geojson)
       target_bbox = target_geom.bounds
       # 获取覆盖目标bbox的Geohash前缀
       gh_prefixes = get_geohash_prefixes(target_bbox, precision=6)
       count = 0
       # 批量查询每个前缀下的多边形
       for prefix in gh_prefixes:
           response = table.query(
               KeyConditionExpression='PK = :pk',
               ExpressionAttributeValues={':pk': prefix}
           )
           for item in response['Items']:
               poly_geom = shape(json.loads(item['geo_json']))
               if poly_geom.intersects(target_geom):
                   count += 1
       return count
   

• 注意事项：
  • Geohash精度需根据数据分布调整：精度越高，分区越细，过滤精度越高，但写入时可能需要生成更多前缀；
  • 对于超大多边形，可拆分多个Geohash前缀覆盖，避免漏查；
  • 可使用DynamoDB的批量查询（batch_get_item）优化多前缀查询的性能。

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