从多REST API抽取数据至不同Kafka主题并去重——工具咨询

Hey there, let's walk through the best tools to handle your use case—pulling data from multiple REST APIs and publishing to corresponding Kafka Topics while avoiding duplicates. These are all tools I've used in production, so I can speak to their pros and cons:

1. Apache Kafka Connect #

This is the go-to choice if you want a native, low-code integration with Kafka. Here's how it works:

• Use the REST Source Connector (or community connectors like confluentinc/kafka-connect-rest) to pull data from each of your APIs. You can configure separate connector instances for different APIs, each targeting a specific Kafka Topic.
• Deduplication:
  • Set a unique identifier from your API response (like a record ID or timestamp) as the Kafka message key using transforms like ExtractField$Key. Kafka's log compaction will automatically keep only the latest record for each key, eliminating duplicates.
  • Enable Kafka's idempotent producers (via enable.idempotence=true in connector config) to guarantee exactly-once delivery, so even if the connector retries, you won't get duplicate messages.
  • For incremental pulls, configure the connector to track a cursor (like an updated_at timestamp) so it only fetches new or modified data each run.

2. Apache NiFi #

NiFi is perfect if you need a visual, flexible ETL pipeline for complex data flows:

• Use the InvokeHTTP processor to call each REST API, then route responses to different Kafka Topics using RouteOnAttribute (based on API source or data type) and PublishKafkaRecord processors.
• Deduplication:
  • Use the DistinctRecord processor to filter out duplicate records based on a unique field (e.g., record_id).
  • For persistent deduplication, use PutDistributedMapCache to store IDs of processed records. Before publishing, use FetchDistributedMapCache to check if the record has already been handled, and drop it if it exists.
  • NiFi's built-in transaction support ensures exactly-once delivery to Kafka, so retries won't create duplicates.

3. Apache Airflow (with Kafka Providers) #

If your API pulls are scheduled (e.g., hourly/daily syncs), Airflow is a great fit:

• Use SimpleHttpOperator to fetch data from each API, then process the response (parse JSON, transform fields) in a Python task. Use KafkaProducerHook to send processed data to the target Kafka Topic.
• Deduplication:
  • Store processed record IDs in a lightweight database (like SQLite, PostgreSQL) or Redis. In your task, query this store before sending to Kafka and skip records that already exist.
  • Enable idempotence in the Kafka producer config (via enable.idempotence=True) to prevent duplicate messages from retries.
  • Airflow's task retry logic can be paired with checkpoints to avoid reprocessing entire datasets.

4. Spring Cloud Stream / Spring Kafka #

For Java/Spring ecosystem projects, this is a robust, code-first approach:

• Create separate bindings for each API-Topic pair using Spring Cloud Stream's declarative model. Use @RestController or RestTemplate to fetch data from APIs, then send messages to the configured Kafka Topic via StreamBridge or KafkaTemplate.
• Deduplication:
  • Use Redis or an in-memory cache to track processed record IDs. Before sending, check if the ID exists and skip if it does.
  • Configure Spring Kafka's producer with enable.idempotence=true to guarantee exactly-once delivery. You can also leverage Kafka's transactional producers for end-to-end exactly-once semantics.

5. Custom Python Scripts (requests + kafka-python) #

If you need a lightweight, quick solution for simple use cases:

• Use the requests library to pull data from your APIs, parse the responses, and use kafka-python to publish messages to the right Topics.
• Deduplication:
  • Maintain a local JSON file or use Redis to store processed record IDs. Check each record against this store before sending.
  • Set the message key to a unique identifier and enable Kafka's idempotent producer to avoid duplicates from retries.
  • Note: You'll need to handle scheduling (e.g., with cron) and error handling manually here.

Final Recommendation #

• Pick Kafka Connect if you want minimal code and native Kafka integration.
• Go with NiFi if you need a visual, flexible pipeline for complex data transformations.
• Use Airflow if your syncs are scheduled and you need workflow orchestration.
• Choose Spring Cloud Stream if you're already in the Java/Spring ecosystem.
• Opt for custom Python scripts for small, simple projects where you want full control.

内容的提问来源于stack exchange，提问作者Justin