elasticsearch - 使用摄取附件批量索引(约40k个.docx类型文件)的嵌套方式是什么？

Question

我对ELK堆栈非常熟悉，目前正在使用Elastic search 6.6。我们的用例是内容搜索约40K .docx文件
(由投资组合经理作为研究报告上载。
允许的最大文件大小为10 MB，但大多数文件大小只有几Kb)。
我已使用摄取附件插件为示例测试文件建立索引，并且还可以使用KIBANA搜索内容
例如:POST / attachment_test / my_type / _search？pretty = true

{
  "query": {
    "match": {
      "attachment.content": "JP Morgan"
    }
  }
}

返回我预期的结果。
我的疑问:

使用摄取插件，我们需要将数据推送到插件。我正在使用VS 2017和 flex NEST dll。这意味着我必须以编程方式读取40K文档，并使用NEST命令将其推送到ES？

我已经完成了Fscrawler项目，并且知道它可以达到目的，但我将其保留为最后的手段

如果我要使用方法1(代码)，是否有任何批量上传API可用于将多个附件一起发布到ES(分批)？

Answer 1

最后，我使用C#代码将40K文件上传到 flex 索引中:

 private static void PopulateIndex(ElasticClient client)
    {
        var directory =System.Configuration.ConfigurationManager.AppSettings["CallReportPath"].ToString();
        var callReportsCollection = Directory.GetFiles(directory, "*.doc"); //this will fetch both doc and docx
        //callReportsCollection.ToList().AddRange(Directory.GetFiles(directory, "*.doc"));
        ConcurrentBag<string> reportsBag = new ConcurrentBag<string>(callReportsCollection);
        int i = 0;
        var callReportElasticDataSet = new DLCallReportSearch().GetCallReportDetailsForElastic();//.AsEnumerable();//.Take(50).CopyToDataTable();
        try
        {
            Parallel.ForEach(reportsBag, callReport =>
            //Array.ForEach(callReportsCollection,callReport=>
            {
                var base64File = Convert.ToBase64String(File.ReadAllBytes(callReport));
                var fileSavedName = callReport.Replace(directory, "");
                // var dt = dLCallReportSearch.GetCallFileName(fileSavedName.Replace("'", "''"));//replace the ' in a file name with '';
                var rows = callReportElasticDataSet.Select("CALL_SAVE_FILE like '%" + fileSavedName.Replace("'", "''") + "'");
                if (rows != null && rows.Count() > 0)
                {
                    var row = rows.FirstOrDefault();
                    //foreach (DataRow row in rows)
                    //{
                    i++;
                    client.Index(new Document
                    {
                        Id = i,
                        DocId = Convert.ToInt32(row["CALL_ID"].ToString()),
                        Path = row["CALL_SAVE_FILE"].ToString().Replace(CallReportPath, ""),
                        Title = row["CALL_FILE"].ToString().Replace(CallReportPath, ""),
                        Author = row["USER_NAME"].ToString(),
                        DateOfMeeting = string.IsNullOrEmpty(row["CALL_DT"].ToString()) ? (DateTime?)null : Convert.ToDateTime(row["CALL_DT"].ToString()),
                        Location = row["CALL_LOCATION"].ToString(),
                        UploadDate = string.IsNullOrEmpty(row["CALL_REPORT_DT"].ToString()) ? (DateTime?)null : Convert.ToDateTime(row["CALL_REPORT_DT"].ToString()),
                        CompanyName = row["COMP_NAME"].ToString(),
                        CompanyId = Convert.ToInt32(row["COMP_ID"].ToString()),
                        Country = row["COU_NAME"].ToString(),
                        CountryCode = row["COU_CD"].ToString(),
                        RegionCode = row["REGION_CODE"].ToString(),
                        RegionName = row["REGION_NAME"].ToString(),
                        SectorCode = row["SECTOR_CD"].ToString(),
                        SectorName = row["SECTOR_NAME"].ToString(),
                        Content = base64File
                    }, p => p.Pipeline("attachments"));
                    //}
                }
            });
        }
        catch (Exception ex)
        {
            throw ex;
        }
    }