欢迎来到尧图网

客户服务 关于我们

您的位置:首页 > 文旅 > 旅游 > 【原创】通过S3接口将海量文件索引导入elasticsearch

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

2025/3/22 11:50:10 来源:https://blog.csdn.net/jxyk2007/article/details/146419479  浏览:    关键词:【原创】通过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

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com

热搜词