过滤 Kafka 记录 Python答案

【问题标题】：Filtering Kafka Records Python过滤 Kafka 记录 Python
【发布时间】：2022-12-07 12:04:45
【问题描述】：

我有一个从不同来源消费的 Kafka 主题，我想根据一些规则过滤记录并将记录传送到两个不同的主题（一个用于有效记录，一个用于无效记录）。知道代码逻辑应该如何吗？谢谢，

【问题讨论】：

【解决方案1】：

这是一个例子。

import os
from kafka import KafkaConsumer, KafkaProducer

# Set up the Kafka consumer
consumer = KafkaConsumer(...)

# Set up the Kafka producers
valid_producer = KafkaProducer(...)
invalid_producer = KafkaProducer(...)

# Implement the filtering rules
def filter_records(record):
  # Replace this with your own filtering logic
  return True if record.value % 2 == 0 else False

# Consume messages from Kafka, filter them, and produce to the appropriate topic
for message in consumer:
  if filter_records(message):
    valid_producer.send('valid-records', message.value)
  else:
    invalid_producer.send('invalid-records', message.value)

【讨论】：