【原创】通过S3接口将海量文件索引导入elasticsearch

在医院海量影像文件通过s3传到蓝光存储时，要找一个文件需要全部文件遍历一遍，效率非常非常低。

S3 是对象存储服务，本身不是专门为快速文件查找设计的，而 Elasticsearch 是搜索引擎，在查找特定文件或数据方面具有明显优势，以下是对这种情况的分析：

S3 查找文件慢的原因

存储结构和查询方式：S3 是一种基于对象的存储系统，它将文件作为对象存储在存储桶中。虽然可以通过文件名或键来检索对象，但这种检索方式相对简单，不支持复杂的查询条件和高效的索引机制。如果要在大量文件中查找特定文件，可能需要遍历整个存储桶或使用前缀匹配等有限的方式，这在文件数量庞大时效率较低。
网络延迟：S3 是基于网络的存储服务，数据传输需要通过网络进行。当请求查找文件时，需要与 S3 服务器进行通信，网络延迟会影响查找速度。特别是在跨地域访问或网络环境不稳定的情况下，网络延迟可能会更加明显，导致查找文件的时间延长。

Elasticsearch 查找文件快的原因

倒排索引：Elasticsearch 使用倒排索引来存储和检索数据。倒排索引将文档中的每个词项映射到包含该词项的文档列表，这使得在查找特定关键词或短语时能够快速定位到相关文档。通过对文件内容进行索引，Elasticsearch 可以在瞬间返回包含特定关键词的文件列表，大大提高了查找效率。
分布式架构和并行处理：Elasticsearch 通常以分布式集群的方式部署，可以将数据分布在多个节点上进行存储和处理。在进行查询时，它可以并行地在多个节点上执行搜索操作，然后将结果合并返回。这种分布式架构和并行处理能力使得 Elasticsearch 能够处理大量的数据，并在短时间内返回查询结果。
丰富的查询功能：Elasticsearch 提供了丰富的查询 DSL（Domain Specific Language），支持各种复杂的查询条件，如全文搜索、精确匹配、范围查询、布尔查询等。用户可以根据自己的需求灵活组合查询条件，快速定位到所需的文件。同时，Elasticsearch 还支持模糊查询、同义词查询等高级功能，进一步提高了查找的准确性和灵活性。
上代码，通过go实现导入索引。

package mainimport ("bytes""context""encoding/json""fmt""github.com/aws/aws-sdk-go-v2/aws""github.com/aws/aws-sdk-go-v2/config""github.com/aws/aws-sdk-go-v2/credentials""github.com/aws/aws-sdk-go-v2/service/s3""github.com/elastic/go-elasticsearch/v8""github.com/elastic/go-elasticsearch/v8/esapi""gopkg.in/ini.v1""log""crypto/tls""net/http""time"
)type S3Config struct {BucketName  stringAccessKey   stringSecretKey   stringEndpointURL string
}type ESConfig struct {Host       stringUser       stringPass       stringIndexName  stringSearchType string
}func readConfig() (S3Config, ESConfig) {cfg, err := ini.Load("config.ini")if err != nil {log.Fatalf("无法读取配置文件: %v", err)}s3Cfg := S3Config{BucketName:  cfg.Section("s3").Key("bucket_name").String(),AccessKey:   cfg.Section("s3").Key("access_key").String(),SecretKey:   cfg.Section("s3").Key("secret_key").String(),EndpointURL: cfg.Section("s3").Key("endpoint_url").String(),}esCfg := ESConfig{Host:       cfg.Section("elasticsearch").Key("host").String(),User:       cfg.Section("elasticsearch").Key("user").String(),Pass:       cfg.Section("elasticsearch").Key("password").String(),IndexName:  cfg.Section("elasticsearch").Key("index_name").String(),SearchType: cfg.Section("elasticsearch").Key("search_type").String(),}return s3Cfg, esCfg
}func getS3ETag(s3Client *s3.Client, bucketName, fileKey string) string {resp, err := s3Client.HeadObject(context.TODO(), &s3.HeadObjectInput{Bucket: aws.String(bucketName),Key:    aws.String(fileKey),})if err != nil {log.Printf("获取 %s 的ETag失败: %v", fileKey, err)return ""}etag := aws.ToString(resp.ETag)if len(etag) > 0 && etag[0] == '"' && etag[len(etag)-1] == '"' {etag = etag[1 : len(etag)-1]}return etag
}func fetchS3Files(s3Client *s3.Client, esClient *elasticsearch.Client, bucketName, indexName string) {paginator := s3.NewListObjectsV2Paginator(s3Client, &s3.ListObjectsV2Input{Bucket: aws.String(bucketName),})for paginator.HasMorePages() {page, err := paginator.NextPage(context.TODO())if err != nil {log.Printf("获取S3文件列表页失败: %v", err)continue}for _, obj := range page.Contents {fileKey := aws.ToString(obj.Key)log.Printf("导入索引：",fileKey)fileSize := aws.ToInt64(obj.Size)lastModified := obj.LastModifiedstorageClass := string(obj.StorageClass) // 修复点etag := getS3ETag(s3Client, bucketName, fileKey)fileData := map[string]interface{}{"file_key":      fileKey,"file_size":     fileSize,"last_modified": lastModified,"storage_class": storageClass,"etag":          etag,}fileDataJSON, err := json.Marshal(fileData)if err != nil {log.Printf("将文件数据转换为JSON失败: %v", err)continue}req := esapi.IndexRequest{Index:   indexName,Body:    bytes.NewReader(fileDataJSON), // 修复点Refresh: "true",}resp, err := req.Do(context.TODO(), esClient)if err != nil {log.Printf("将文件数据索引到Elasticsearch失败: %v", err)continue}defer resp.Body.Close()}}fmt.Println("S3 文件索引完成")
}func main() {s3Cfg, esCfg := readConfig()customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {return aws.Endpoint{URL:               s3Cfg.EndpointURL,SigningRegion:     "us-east-1", // 替换为你的实际regionHostnameImmutable: true,}, nil})awsCfg, err := config.LoadDefaultConfig(context.TODO(),config.WithRegion("us-east-1"), // 替换为你的实际regionconfig.WithEndpointResolverWithOptions(customResolver),config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(s3Cfg.AccessKey,s3Cfg.SecretKey,"",)),)if err != nil {log.Fatalf("无法加载S3配置: %v", err)}s3Client := s3.NewFromConfig(awsCfg)esCfgOptions := elasticsearch.Config{Addresses: []string{esCfg.Host},Username:  esCfg.User,Password:  esCfg.Pass,Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true, // ⚠️ 跳过证书校验（不安全，仅限开发）},ResponseHeaderTimeout: 10 * time.Second,},}esClient, err := elasticsearch.NewClient(esCfgOptions)if err != nil {log.Fatalf("无法创建Elasticsearch客户端: %v", err)}fetchS3Files(s3Client, esClient, s3Cfg.BucketName, esCfg.IndexName)
}

config.ini文件

[elasticsearch]
host = https://localhost:9200
user = elastic
password = UfI*****uq
index_name = test
search_type = wildcard[s3]
bucket_name = test
access_key = V4H***RPD6DB
secret_key = lHdm*********n9UjlS
endpoint_url = http://172.*.*.18:7480/
connect_timeout = 10