.NET Core 2下无需Logstash将DataTable/CSV导入Elasticsearch的方法

嘿，我之前也碰到过这个问题——不想靠Logstash的自定义配置文件来处理每个CSV，直接用.NET Core 2搞定CSV或DataTable上传到Elasticsearch，下面给你几个亲测有效的方案：

NEST是Elasticsearch官方提供的强类型.NET客户端，用起来很顺手，适配.NET Core 2.x：

• 第一步，安装对应版本的NuGet包（注意：客户端版本必须和你的Elasticsearch服务器版本一致，比如服务器是6.8.x，就装6.8.x的NEST）：

  Install-Package NEST -Version 6.8.10
  

• 第二步，用CsvHelper读取CSV（轻量高效的CSV处理库，同样选适配.NET Core 2的版本）：

  Install-Package CsvHelper -Version 7.1.1
  

• 第三步，编写上传代码：
  先定义和CSV列对应的实体类，比如你的CSV有Id、Username、RegisterDate三列：

  public class UserCsvModel
  {
      public int Id { get; set; }
      public string Username { get; set; }
      public DateTime RegisterDate { get; set; }
  }
  

  然后读取CSV并批量上传：

  using CsvHelper;
  using Nest;
  using System.IO;
  using System.Linq;
  
  // 初始化Elasticsearch客户端
  var esSettings = new ConnectionSettings(new Uri("http://你的ES地址:9200"))
      .DefaultIndex("user-data-index"); // 默认索引名
  var esClient = new ElasticClient(esSettings);
  
  // 读取CSV文件
  using (var streamReader = new StreamReader(@"D:\user_data.csv"))
  using (var csvReader = new CsvReader(streamReader))
  {
      // 把CSV行转成实体列表
      var userRecords = csvReader.GetRecords<UserCsvModel>().ToList();
  
      // 批量上传，等待索引刷新确保数据可见
      var bulkResponse = esClient.Bulk(b => b
          .IndexMany(userRecords)
          .Refresh(Refresh.WaitFor));
  
      // 处理上传错误
      if (!bulkResponse.IsValid)
      {
          foreach (var errorItem in bulkResponse.ItemsWithErrors)
          {
              Console.WriteLine($"上传文档{errorItem.Id}失败：{errorItem.Error.Reason}");
          }
      }
  }
  

如果你的数据已经是DataTable格式，不用转实体类的话，可以用Elasticsearch.Net（低级客户端，更灵活）：

• 先安装NuGet包：

  Install-Package Elasticsearch.Net -Version 6.8.10
  

• 编写上传代码：

  using Elasticsearch.Net;
  using System.Data;
  using System.Collections.Generic;
  using System.Linq;
  
  // 初始化低级客户端
  var esNode = new Uri("http://你的ES地址:9200");
  var esConfig = new ConnectionConfiguration(esNode);
  var lowLevelClient = new ElasticLowLevelClient(esConfig);
  
  // 构造批量请求内容
  var bulkRequests = new List<object>();
  foreach (DataRow row in yourDataTable.Rows)
  {
      // 添加索引元数据（指定索引名和文档ID）
      bulkRequests.Add(new { index = new { _index = "your-data-index", _id = row["Id"].ToString() } });
      // 把DataRow转成字典作为文档内容
      var docContent = row.Table.Columns.Cast<DataColumn>()
          .ToDictionary(col => col.ColumnName, col => row[col]);
      bulkRequests.Add(docContent);
  }
  
  // 发送批量请求
  var bulkResponse = lowLevelClient.Bulk<DynamicResponse>(bulkRequests);
  
  // 错误处理
  if (!bulkResponse.Success)
  {
      Console.WriteLine($"批量上传失败：{bulkResponse.ServerError?.Error?.Reason}");
  }
  

• 版本匹配：一定要保证Elasticsearch客户端版本和服务器版本完全一致，否则会出现各种兼容性问题（.NET Core 2.x建议搭配Elasticsearch 6.x系列）。
• 分批上传：如果数据量很大（比如几万条以上），不要一次性上传所有数据，分成每次1000-5000条的小批量，避免内存溢出和请求超时。
• 自定义映射：如果需要指定字段类型（比如日期、整数、关键字类型），最好提前创建索引映射，避免Elasticsearch自动推断的类型不符合预期。比如创建映射的代码：

  var createIndexResponse = esClient.Indices.Create("user-data-index", c => c
      .Map<UserCsvModel>(m => m
          .Properties(p => p
              .Date(d => d.Name(n => n.RegisterDate).Format("yyyy-MM-dd"))
              .Keyword(k => k.Name(n => n.Username)))));
  

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