Kafka作为消息队列接收流式数据，Cassandra存储高写入的海量数据，Elasticsearch用于实时搜索和分析。通过Kafka Producer发送数据到Kafka Topic，Kafka Consumer消费数据写入Cassandra，同时将数据推送到Elasticsearch Index，实现流式数据中心。配置Kafka Connect Sink Connector将数据从Kafka同步到Cassandra和Elasticsearch。

架构设计

典型架构：数据源 -> Kafka -> (Cassandra Sink + Elasticsearch Sink)。使用Kafka Streams或Flink处理数据流，聚合后持久化到Cassandra，索引到Elasticsearch。Kafka确保数据不丢失，Cassandra处理高吞吐，Elasticsearch提供毫秒级查询。

部署步骤

1. 启动Kafka集群，创建Topic如"stream-data"。2. 配置Cassandra集群，创建Keyspace和Table。3. 部署Elasticsearch集群，创建Index。4. 使用Kafka Connect部署Cassandra Sink Connector和Elasticsearch Sink Connector。5. 编写Producer代码：Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); KafkaProducer producer = new KafkaProducer<>(props); producer.send(new ProducerRecord("stream-data", key, value))。6. 测试数据流：发送数据到Kafka，验证Cassandra和Elasticsearch收到。

实际案例

在日志系统中，Flume或Logstash将日志推到Kafka，Kafka Consumer批量写入Cassandra表中，同时通过Kafka Connect Elasticsearch插件将JSON日志索引到ES，实现实时监控和搜索。

配置示例

Cassandra Sink Connector config: {"connector.class":"io.confluent.connect.cassandra.CassandraSinkConnector","topics":"stream-data","cassandra.contact.points":"cassandra-host","keyspace":"stream_keyspace"}。Elasticsearch Sink: {"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","topics":"stream-data","connection.url":"http://es-host:9200","type.name":"_doc"}。

优化建议

使用Kafka的Exactly-Once语义，避免数据重复。Cassandra用TimeUUID分区键支持时间序列。Elasticsearch用Logstash或Beats管道优化索引。监控工具如Prometheus + Grafana观察整个流。

代码片段

Consumer代码：@KafkaListener(topics = "stream-data") public void consume(String message) { // 解析JSON，写入Cassandra session.execute(insertStmt.bind(values)); // 索引到ES elasticsearchClient.index(indexRequest); }

FAQ
Q: Kafka、Cassandra和Elasticsearch哪个负责存储？
A: Cassandra负责持久化存储，Elasticsearch负责搜索索引，Kafka负责传输。
Q: 如何处理数据丢失？
A: 配置Kafka replication factor >1，启用acks=all。
Q: 性能瓶颈在哪里？
A: 通常在Consumer处理速度，优化batch size和线程池。
Q: 支持多租户吗？
A: 是，用Kafka Topic分区、Cassandra tenant keyspace、ES index per tenant。