Azure Cosmos DB Gremlin经纬度格式转换问题：机场数据转顶点

最近我在做一个基于Azure Cosmos DB Graph API的项目，需要把一份包含6000多条记录的机场地理坐标数据（格式是[IATA代码, 纬度, 经度]）转换成Gremlin顶点。我选用了ASP.NET Core 2.0控制台程序来实现，借助CSVReader流式处理airport.dat这个CSV文件，整个过程踩了一些小坑，这里把实现思路和代码整理出来，希望能帮到有同样需求的朋友。

一、前期准备 #

• 确保你的ASP.NET Core 2.0控制台项目已经引用了必要的NuGet包：
  • CsvHelper：用来高效读取CSV文件，支持流式处理避免内存过载
  • Gremlin.Net：用来和Azure Cosmos DB Graph API建立交互
• 提前确认airport.dat的文件格式，比如蒙特利尔皮埃尔·埃利奥特·特鲁多国际机场的记录示例：

"Montreal-Pierre Elliott Trudeau International Airport","YUL","Montreal","Canada",45.4706,-73.7407,118, "America/Toronto","airport","IATA"

二、CSV流式读取处理 #

因为数据量有6000多行，用流式读取可以避免一次性把所有数据加载到内存，提升程序运行效率。我定义了对应的实体类来映射CSV列，再通过CsvHelper逐行读取：

using CsvHelper;
using System.IO;
using System.Threading.Tasks;

// 定义机场数据实体类，对应CSV中的核心字段
public class AirportData
{
    public string Name { get; set; }
    public string IataCode { get; set; }
    public string City { get; set; }
    public string Country { get; set; }
    public double Latitude { get; set; }
    public double Longitude { get; set; }
}

// CSV列映射配置，根据airport.dat的列顺序调整索引
public sealed class AirportDataMap : ClassMap<AirportData>
{
    public AirportDataMap()
    {
        Map(m => m.Name).Index(0);
        Map(m => m.IataCode).Index(1);
        Map(m => m.City).Index(2);
        Map(m => m.Country).Index(3);
        Map(m => m.Latitude).Index(4);
        Map(m => m.Longitude).Index(5);
    }
}

// 流式读取CSV文件的核心逻辑
public async Task ProcessAirportCsv()
{
    using (var reader = new StreamReader("airport.dat"))
    using (var csv = new CsvReader(reader))
    {
        csv.Configuration.RegisterClassMap<AirportDataMap>();
        
        // 逐行读取，避免内存溢出
        while (csv.Read())
        {
            var airport = csv.GetRecord<AirportData>();
            // 调用顶点转换方法
            await ConvertToGremlinVertex(airport);
        }
    }
}

三、转换为Gremlin顶点并写入Cosmos DB #

接下来要把读取到的机场数据转换成Gremlin顶点，再写入Azure Cosmos DB。这里建议做批量处理优化，避免单条请求频繁调用影响性能：

using Gremlin.Net.Driver;
using Gremlin.Net.Driver.Remote;
using Gremlin.Net.Structure.IO.GraphSON;

private static async Task ConvertToGremlinVertex(AirportData airport)
{
    // 初始化Gremlin客户端，替换成你的Cosmos DB信息
    var gremlinServer = new GremlinServer(
        "your-cosmosdb-account.gremlin.cosmos.azure.com",
        443,
        enableSsl: true,
        username: "/dbs/your-database/colls/your-graph-collection",
        password: "your-cosmosdb-primary-key");
    
    using (var client = new GremlinClient(gremlinServer, new GraphSON2Reader(), new GraphSON2Writer(), GremlinClient.GraphSON2MimeType))
    {
        // 构建Gremlin创建顶点的查询，用IATA代码作为顶点唯一ID
        var gremlinQuery = $"g.addV('airport')" +
                           $".property('id', '{airport.IataCode}')" +
                           $".property('name', '{airport.Name}')" +
                           $".property('city', '{airport.City}')" +
                           $".property('country', '{airport.Country}')" +
                           $".property('latitude', {airport.Latitude})" +
                           $".property('longitude', {airport.Longitude})";
        
        await client.SubmitAsync<dynamic>(gremlinQuery);
    }
}

四、关键优化与注意事项 #

• 批量处理：6000条数据逐个写入效率极低，建议每50-100条数据组成一个批量Gremlin请求，减少网络交互次数
• 数据校验：处理前要校验纬度、经度的数值合法性，过滤掉无效数据避免写入错误
• 索引配置：在Cosmos DB中为顶点的id字段（即IATA代码）配置唯一索引，防止重复创建顶点
• 错误重试：添加try-catch块处理Cosmos DB的限流（429错误），实现指数退避重试机制，保证数据写入成功率

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