展开全文

因此，我们决定构建一个基于事件队列的基础设施。还因为我们已经计划了一些适合基于事件的未来场景和服务，比如通知服务、数据仓库、微服务架构等。事不宜迟，让我们直接开始解决方案及所使用服务的基本介绍吧。

服务简介

为了实现基于事件的流基础设施，我们决定使用 Confluent Kafka 技术栈。

以下是我们整合的服务:

来源：Confluent公司

Apache Kafka：Kafka 是 Confluent 平台的核心。它是一个基于开源的分布式事件流平台。它将是数据库事件（插入、更新和删除）的主存储区域。

Kafka Connect：我们使用 Kafka-Connect 从 Debezium的 Postgres 连接器获取 Kafka 的数据，该连接器从 Postgres WAL文件中获取事件。

在接收端，我们使用 ElasticSearch 连接器处理数据并将其加载到 ElasticSearch 中。Connect 既可以作为一个独立软件运行，也可以作为一个生产环境容错且可伸缩的服务运行。

ksqlDB：ksqlDB 允许在 Kafka 之上构建一个流处理应用程序。它在内部使用 Kafka-streams 并在事件进来时进行转换，我们使用它来丰富特定流的事件，其中包括已经在 Kafka 持久存在的其他表的事件，这些事件可能与搜索功能相关，例如 root 表中的tenant_id。

自作者的图片：基于 Apache Kafka 的 ksqlDB

使用 ksqlDB，只需编写 SQL 查询来过滤、聚合、关联和填充数据即可。例如，假设我们正在接收一个关于两个主题的事件流，其中包括与brands和brand_products相关的信息。考虑到这是一个多租户数据源，我们需要使用 tenant_id 来填充 brand_product，而 tenant_id目前只与brands相关联。然后，我们可以使用这些填充后的记录，并将它们以非标准化的形式保存在 Elasticsearch 中（以便进行搜索）。

我们可以使用一个主题来设置 KStream:

为了只使用其中几列并按 id 对数据流分区，我们可以创建一个名为 enriched_brands 的新数据流：

然后可以通过 KTable 中的最新偏移量来实现事件集合。我们使用这个功能是为了将brand事件的当前状态与其他流关联起来。

现在我们添加了一个含有brand_id 字段的 brand_products 的新流，但没有tenant_id 字段。

我们可以使用以下关联查询向 brand_products填充 tenant_id。

Schema 注册表：它在 Kafka 的上层，用于存储你在 Kafka 中提取的事件的元数据。它基于 AVRO 模式，并提供 REST 接口来存储和查询它们。它有助于确保一些 Schema 兼容性检查及其随时间发生的演变。

配置技术栈

我们使用 Docker 和 Docker-compose 来配置和部署服务。下面是准备用于构建服务所写的 docker-compose 文件，将运行 Postgres，Elasticsearch，和 Kafka 相关的服务。下面我还将解释提到的每一种服务。

Postgres 和 Elasticsearch

用于 Postgres 的 Docker-compose 服务

用于 Elasticsearch 的 Docker-compose 服务

为了从源数据库中流式的导出事件，我们需要启用逻辑解码以便从其日志中进行复制。在 Postgres 的例子中，这些日志被称为 Write-Ahead Logs (WAL) ，它们被写入一个文件中。我们需要一个逻辑解码插件，在我们的例子中，wal2json 用来提取关于持久数据库更改的易于阅读的信息，以便它可以被作为事件发送到 Kafka。

为了配置所需的扩展，你可以参考这个 Postgres Dockerfile文件。

对于 Elasticsearch 和 Postgres，我们需要在环境文件中指定一些必要的变量来设置它们，如用户名、密码等。

Zookeeper

总的来说，Zookeeper 扮演 Kafka 这样的分布式平台的中心服务，它存储所有元数据，如 Kafka 节点状态，并持续跟踪主题或分区。

即便已经有了在无 zookeeper 的情况下运行 Kafka的替代计划，但是目前它还是管理集群所必须的。

Kafka Broker

为了简单起见，我们将配置一个单节点 Kafka 集群。我将在本系列的第 2 部分中讨论关于多阶段集群的更多内容。

了解我们为 Kafka Broker 所做的一些配置尤其重要。

监听器（Listeners）

因为 Kafka 被设计成一个分布式平台，我们需要提供一些明确的方式来允许 Kafka Broker 彼此在内部通信，并基于您的网络结构与其他客户端进行外部通信。因此我们使用监听器来完成这个任务，监听器是主机、端口和协议的组合。

• KAFKA_LISTENERS

这是一个可以由 KAFKA 绑定的网络端口列表，由主机、端口和协议组合成。默认情况下，它被设置为 0.0.0.0，即监听所有端口。

• KAFKA_ADVERTISED_LISTENERS

这个值同样是主机和端口的组合，客户端将使用它来连接 KAFKA Broker。因此，如果客户端在 docker 中，它可以使用 broker:9092连接到 broker，如果在 docker 外，则返回 localhost:9092来建立和 broker 的连接。我们还需要提到监听器名称，其才能被映射到恰当的协议以建立连接。

• KAFKA_LISTENER_SECURITY_PROTOCOL_MAP

这里我们将用户定义的监听器名称映射到希望用于通信的协议；它可以是PLAINTEXT（未加密）或 SSL (加密的)。这些名字在 KAFKA_LISTENERS 和 KAFKA_ADVERTISED_LISTENERS 中被进一步与 host/ip 一起使用，以便使用恰当的协议。

由于我们只配置了单节点的 Kafka 集群，因此返回的或者说发送给任何客户端的推荐地址都将是自身这同一 broker。

Schema 注册（Schema-Registry）

对于单节点 schema 注册，我们指定用来连接 zookeeper 的字符串，Kafka 用它存储与 schema 相关的数据。

Schema 注册（Schema-Registry）

我们看到一些新的参数，比如:

• CONNECT_BOOTSTRAP_SERVERS：一组主机和端口组合，用于建立到 Kafka 集群的初始连接
• CONNECT_KEY_CONVERTER：用于将键（key）从connect格式序列化为与 Kafka 兼容的格式。类似地，对于 CONNECT_VALUE_CONVERTER，我们使用 AvroConverter 进行序列化。

映射大量 source 和 sink 连接器插件并在 CONNECT_PLUGIN_PATH 中指定它们是非常的重要。

Kafka-Connect

如果不打算使用 Kafka-Connect，并且不需要独立于 ksql扩展 Kafka-Connect，那么可以为 ksql设置 embedded-connect配置，这将暴露来自 ksqldb-server的连接点。

除此之外，还有一个环境变量需要考虑:

• KSQL_KSQL_STREAMS_MAX_TASK_IDLE_MS：在当前版本的 ksqlDB，对于流式表关联，关联的结果可能变成不确定的，即如果在流事件之前还没有创建或更新被关联的表中的实时事件，那您可能无法关联成功。当流中的某个事件在某个特定时间戳到达时，配置这个环境变量可以做一些等待让这个事件加载到表中。这提高了关联的可预测性，但可能会导致某些性能下降。在 这里 我们正在努力改善这一点。

实际上，如果你不能清楚地理解上面的内容，我建议你现在就使用这个配置，因为它很有效；它实际上需要另一篇文章来详细讨论时间同步，或者如果你仍然好奇，你可以观看这个由来自 Confluent 的 Matthias j. Sax 制作的视频。

在测试或开发环境中，使用 ksqldb-cli服务来尝试和测试流非常方便。即使在生产环境中，如果您想探索事件流或 Ktables，或者手动创建或过滤流，也可以这样做。尽管如此，还是建议您使用 ksql 或 kafka 客户端或其 REST 端点自动创建流、表或主题，这些我们将在下面进行讨论。

图片由作者提供：目前为止对我们的架构进行的更详细观察

初始化数据

流

这个服务的目的是进行流初始化和 Kafka 内部配置，以及我们正在使用的其他服务。在部署时，我们不希望在服务器上手动创建主题、流、连接等。因此，我们使用为每个服务提供的 REST 服务，并编写 shell 脚本来自动化这个过程。

我们的配置脚本如下所示:

# Setup ENV variables in connectors json filessed -i “s/POSTGRES_USER/${POSTGRES_USER}/g” connectors/postgres.jsonsed -i “s/POSTGRES_PASSWORD/${POSTGRES_PASSWORD}/g” connectors/postgres.jsonsed -i “s/POSTGRES_DB/${POSTGRES_DB}/g” connectors/postgres.jsonsed -i “s/ELASTIC_PASSWORD/${ELASTIC_PASSWORD}/g” connectors/elasticsearch.json

# Simply wait until original kafka container and zookeeper are started.export WAIT_HOSTS=zookeeper:2181,broker:9092,schema-registry:8081,ksqldb-server:8088,elasticsearch:9200,connect:8083export WAIT_HOSTS_TIMEOUT=300/wait

# Parse string of kafka topics into an array# https://stackoverflow.com/a/10586169/4587961kafkatopicsArrayString=”$KAFKA_TOPICS”IFS=’, ‘ read -r -a kafkaTopicsArray <<< “$kafkatopicsArrayString”

# A separate variable for zookeeper hosts.zookeeperHostsValue=$ZOOKEEPER_HOSTS

# Terminate all queriescurl -s -X “POST” “http://ksqldb-server:8088/ksql” -H “Content-Type: application/vnd.ksql.v1+json; charset=utf-8” -d ‘{“ksql”: “SHOW QUERIES;”}’ | jq ‘.[].queries[].id’ | xargs -Ifoo curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Content-Type: application/vnd.ksql.v1+json; charset=utf-8” -d ‘{“ksql”: “TERMINATE ‘foo’;”}’

# Drop All Tablescurl -s -X “POST” “http://ksqldb-server:8088/ksql” -H “Content-Type: application/vnd.ksql.v1+json; charset=utf-8” -d ‘{“ksql”: “SHOW TABLES;”}’ | jq ‘.[].tables[].name’ | xargs -Ifoo curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Content-Type: application/vnd.ksql.v1+json; charset=utf-8” -d ‘{“ksql”: “DROP TABLE “foo”;”}’

# Drop All Streamscurl -s -X “POST” “http://ksqldb-server:8088/ksql” -H “Content-Type: application/vnd.ksql.v1+json; charset=utf-8” -d ‘{“ksql”: “SHOW STREAMS;”}’ | jq ‘.[].streams[].name’ | xargs -Ifoo curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Content-Type: application/vnd.ksql.v1+json; charset=utf-8” -d ‘{“ksql”: “DROP STREAM “foo”;”}’

# Create kafka topic for each topic item from split array of topics.for newTopic in “${kafkaTopicsArray[@]}”; do # https://kafka.apache.org/quickstart curl -X DELETE http://elasticsearch:9200/enriched_$newTopic –user elastic:${ELASTIC_PASSWORD} curl -X DELETE http://schema-registry:8081/subjects/store.public.$newTopic-value kafka-topics –create –topic “store.public.$newTopic” –partitions 1 –replication-factor 1 –if-not-exists –zookeeper “$zookeeperHostsValue” curl -X POST -H “Content-Type: application/vnd.schemaregistry.v1+json” –data @schemas/$newTopic.json http://schema-registry:8081/subjects/store.public.$newTopic-value/versions

done

curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Accept: application/vnd.ksql.v1+json” -d { “ksql”: “CREATE STREAM \”brands\” WITH (kafka_topic = ‘store.public.brands’, value_format = ‘avro’);”, “streamsProperties”: {} }’curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Accept: application/vnd.ksql.v1+json” -d { “ksql”: “CREATE STREAM \”enriched_brands\” WITH ( kafka_topic = ‘enriched_brands’ ) AS SELECT CAST(brand.id AS VARCHAR) as \”id\”, brand.tenant_id as \”tenant_id\”, brand.name as \”name\” from \”brands\” brand partition by CAST(brand.id AS VARCHAR) EMIT CHANGES;”, “streamsProperties”: {} }’

curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Accept: application/vnd.ksql.v1+json” -d { “ksql”: “CREATE STREAM \”brand_products\” WITH ( kafka_topic = ‘store.public.brand_products’, value_format = ‘avro’ );”, “streamsProperties”: {} }’curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Accept: application/vnd.ksql.v1+json” -d { “ksql”: “CREATE TABLE \”brands_table\” AS SELECT id as \”id\”, latest_by_offset(tenant_id) as \”tenant_id\” FROM \”brands\” group by id EMIT CHANGES;”, “streamsProperties”: {} }’curl -X “POST” “http://ksqldb-server:8088/ksql” -H “Accept: application/vnd.ksql.v1+json” -d { “ksql”: “CREATE STREAM \”enriched_brand_products\” WITH ( kafka_topic = ‘enriched_brand_products’ ) AS SELECT \”brand\”.\”id\” as \”brand_id\”, \”brand\”.\”tenant_id\” as \”tenant_id\”, CAST(brand_product.id AS VARCHAR) as \”id\”, brand_product.name AS \”name\” FROM \”brand_products\” AS brand_product INNER JOIN \”brands_table\” \”brand\” ON brand_product.brand_id = \”brand\”.\”id\” partition by CAST(brand_product.id AS VARCHAR) EMIT CHANGES;”, “streamsProperties”: {} }’

curl -X DELETE http://connect:8083/connectors/enriched_writercurl -X “POST” -H “Content-Type: application/json” –data @connectors/elasticsearch.json http://connect:8083/connectors

curl -X DELETE http://connect:8083/connectors/event_readercurl -X “POST” -H “Content-Type: application/json” –data @connectors/postgres.json http://connect:80

这就是我们目前的工作方式：

• 在运行任何任务之前，我们确保所有的服务都准备好了；
• 我们需要确保主题在 Kafka 上已存在，或者我们创建新的主题；
• 即使有 schema 更新，我们的数据流也应该是可用的；
• 当底层数据 srouce 或 sink 的密码或版本更改，需要再次创建连接。

共享这个配置脚本的目的只是为了演示一种自动化这些 pipeline 的方法。完全相同的配置可能并不适合您，但是自动化工作流和避免在任何环境中的进行手工部署的想法始终是一样的。

为了让这个数据基础设施能够真正快速地运行起来，请参考 Github 仓库:

behindthescenes-group/oesophagus

https://github.com/behindthescenes-group/oesophagus

在你的终端中克隆代码库并执行以下操作:

在 Postgres 数据库 store中创建 brands 和 brand_products 表：

在brands表中插入一些记录：

然后brand_products表中的一些记录:

在 Elasticsearch 的中查看填充了tenant_id 的brand_products ：

我将持续为上述代码库做出贡献：添加在 Kubernetes 部署多节点 Kafka 基础设施的配置，编写更多连接器，使用期望的服务实现即插即用架构的框架。请在这里自由的提交贡献，或让我知道在你在当前配置中所遇到的任何数据工程问题。

下一步

我希望这篇文章能给你一个关于部署和运行完整 Kafka 技术栈的清晰思路，这是一个构建实时流处理应用程序的基础且有效的示例。

根据产品或公司的自身特点，部署过程根据需要可能会有所不同。我还计划在本系列的下一部分中就这样一个系统在可伸缩性方面进行探讨，那将是关于在相同使用场景下如何在 Kubernetes 上部署这样的基础设施的讨论。

英文原文链接：

https://towardsdatascience.com/enabling-a-powerful-search-capability-building-and-deploying-a-real-time-stream-processing-etl-a27ecb0ab0ae

市值超万亿港元！那个CEO写下70万行代码的公司上市了

天下没有免费的App

深圳南山法院：微信好友关系不属于用户隐私；任正非就注册姚安娜商标事件道歉；基金成为年终奖投资首选 | Q资讯

InfoQ 写作平台欢迎所有热爱技术、热爱创作、热爱分享的内容创作者入驻！

还有更多 超值活动等你来！

填写申请，成为作者

开启你的创作之路吧~

点个在看少个 bug👇