可扩展的数据摄取是OpenSearch等大规模分布式搜索和分析引擎的一个关键方面。

构建实时数据摄取管道的方法之一是使用Apache

Kafka

可扩展的数据摄取是OpenSearch等大规模分布式搜索和分析引擎的一个关键方面。

构建实时数据摄取管道的方法之一是使用Apache

Kafka。

它是一个开源事件流平台用于处理高数据量和速度并与包括关系数据库和

NoSQL

数据库在内的各种来源集成。

例如规范用例之一是异构系统源组件之间的数据实时同步以确保

OpenSearch

索引是最新的并且可以通过仪表板和可视化用于分析或使用下游应用程序。

这篇博文将介绍如何创建数据管道其中写入

结合使用则可以使用该连接器。

但有时有特定的要求或原因可能需要使用定制解决方案。

Kafka

不支持的数据源很少见但可能会发生并且不想从头开始编写数据源。

或者这可能是一次性集成您想知道是否值得花费精力来设置和配置

Kafka

提供了多种编程语言的客户端库使您可以编写自己的集成层。

这正是本博客所涵盖的内容我们将利用自定义Go应用程序通过Kafka和OpenSearch的

你将学习

是一个开源搜索和分析引擎用于日志分析、实时监控和点击流分析。

Amazon

OpenSearch

7.10该软件的最终开源版本。

创建集群时您可以选择使用哪个搜索引擎。

OpenSearch

服务提供按需无服务器配置来消除操作复杂性。

它使用索引集合来支持特定的工作负载与传统集群不同它分离了索引和搜索组件并使用Amazon

OpenSearch

处理流数据。

它处理集群管理操作例如创建、更新和删除。

您可以使用标准

Apache

中的一种集群类型无需手动管理和扩展集群容量。

它根据需求自动配置和扩展资源并负责主题分区管理。

采用即用即付定价模式您只需为实际使用量付费。

MSK

Serverless

值得注意的是该博客文章已针对简单性和易于理解进行了优化因此该解决方案并未针对运行生产工作负载进行调整。

以下是一些已进行的简化

MSK

主题的数据。

但是您可以尝试运行使用者应用程序的多个实例并查看数据如何在实例之间分布。

不是使用

Kafka

根据数据量和可扩展性要求为您的消费者应用程序选择合适的计算平台

OpenSearch

身份验证。

以下是消费者应用程序的片段展示了它在实践中的工作原理

func

config.LoadDefaultConfig(context.Background(),

config.WithCredentialsProvider(ec2rolecreds.New()))creds,

err

cfg.Credentials.Retrieve(context.Background())

//....func

[]kgo.Opt{kgo.SeedBrokers(strings.Split(mskBroker,

,)...),kgo.SASL(sasl_aws.ManagedStreamingIAM(func(ctx

context.Context)

creds.SecretAccessKey,SessionToken:

nil})),

无服务器集群。

执行此操作后记下以下集群信息VPC、子网、安全组“属性”选项卡和集群端点单击“查看客户端信息”。

应用程序

cluster,arn:aws:kafka:us-east-1:AWS_ACCOUNT_ID:topic/MSK_CLUSTER_NAME/*,arn:aws:kafka:us-east-1:AWS_ACCOUNT_ID:group/MSK_CLUSTER_NAME/*]},{Effect:

Allow,Action:

注意出于本教程的目的我们选择公共访问类型。

建议为生产工作负载选择VPC。

AWS

https://github.com/build-on-aws/opensearch-using-kafka-golangcd

启动生产者应用程序

脚本该脚本将调用HTTP您刚刚启动的应用程序公开的端点并使用movies.txt以下格式提交电影数据来自文件JSONcurl

./send-data.sh

http://ia.media-imdb.com/images/M/MV5BMTQxNTc3NDM2MF5BMl5BanBnXkFtZTcwNzQ5NTQ3OQ._V1_SX400_.jpg,

plot:

http://ia.media-imdb.com/images/M/MV5BMTQyMDE0MTY0OV5BMl5BanBnXkFtZTcwMjI2OTI0OQ._V1_SX400_.jpg,

plot:

当生产者应用程序忙于向movies主题发送数据时您可以启动消费者应用程序开始处理来自

MSK

OPENSEARCH_INDEX_NAMEmovies-index

export

http://ia.media-imdb.com/images/M/MV5BMTQxNTc3NDM2MF5BMl5BanBnXkFtZTcwNzQ5NTQ3OQ._V1_SX400_.jpg,

plot:

http://ia.media-imdb.com/images/M/MV5BMTQyMDE0MTY0OV5BMl5BanBnXkFtZTcwMjI2OTI0OQ._V1_SX400_.jpg,

plot:

集合中为电影建立索引。

不过您不必等待它完成。

一旦有了数百条记录您就可以继续导航到OpenSearch

OpenSearch

让我们从一个简单的查询开始列出索引中的所有文档不带任何参数或过滤器。

GET

对象。

使用该_source选项从选定字段检索源。

例如要仅检索title、plot和genres字段请运行以下查询

GET

您可以使用术语查询来实现此目的。

例如要搜索字段christmas中包含该术语的电影title请运行以下查询

GET

使用聚合根据特定字段中的值分组来计算汇总值。

例如您可以汇总ratings、

、genre和

等字段year以根据这些字段的值搜索结果。

通过聚合我们可以回答这样的问题“每种类型有多少部电影”

GET

中然后以不同的方式对其进行查询。

在此过程中您还了解了生产工作负载需要记住的架构注意事项和计算选项以及使用基于

Kafka