stream-processing

流处理

用途

此技能可使用 Kafka、Flink 和 Apache Spark 等框架实时处理连续数据流。它适用于需要立即摄取、转换和分析数据的场景，以支持数据工程流水线。

使用时机

当 IoT 传感器、日志文件或金融交易等高容量数据源需要实时分析时，请使用此技能。当批处理无法满足需求时，例如监控系统指标、检测异常或动态更新仪表板，也可使用此技能。

主要能力

• 借助 Kafka 的分布式架构处理高吞吐量数据流，并支持主题、分区和副本，以实现容错。
• 在 Flink 中执行有状态计算，使用窗口（例如，用于 1 分钟聚合的滚动窗口）和精确一次处理语义。
• 集成 Apache Spark Streaming，实现可扩展处理，并利用 DStreams 或 Structured Streaming API 执行 map 和 reduce 等转换。
• 支持背压处理以防止过载，例如使用 Flink 可配置的检查点间隔。

使用模式

• 生产者-消费者模式：通过 Kafka 生产者摄取数据，并由 Flink 消费者处理。例如，将日志发送到 Kafka 主题，并使用 Flink 实时筛选和聚合日志。
• 窗口聚合：在 Flink 中应用基于时间的窗口来汇总数据，例如统计每分钟的事件数。
• ETL 流水线：使用 Spark Streaming 从 Kafka 提取数据，通过 SQL 查询进行转换，并加载到 Elasticsearch 等数据库。
• 容错处理：在 Flink 作业中配置检查点，以便在故障后恢复，确保生产环境中不会丢失数据。

常用命令/API

• Kafka CLI 命令：使用 kafka-console-producer --topic my-topic --broker-list localhost:9092 发送消息。消费消息时，使用：kafka-console-consumer --topic my-topic --from-beginning --bootstrap-server localhost:9092。
• Flink 命令：使用 flink run -c com.example.StreamJob /path/to/jar --input kafka-topic --output file:///output 提交作业以处理数据流。使用 Flink 的 REST API http://localhost:8081/jobs/overview 进行监控。
• Spark Streaming API：在 Scala 中，使用 val stream = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").load() 创建数据流。然后应用转换：stream.selectExpr("CAST(value AS STRING)").writeStream.outputMode("append").format("console").start()。
• 配置格式：Kafka 需要使用 properties 文件，例如为生产者配置 key.serializer=org.apache.kafka.common.serialization.StringSerializer。Flink 使用 YAML 配置，例如 execution.checkpointing.interval: 1min。

集成说明

要将 Kafka 集成为 Flink 的数据源，请在 Flink 作业中添加依赖项（例如，通过 Maven：<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId></dependency>）。进行身份验证时，在生产者脚本中设置 $KAFKA_API_KEY 等环境变量：export KAFKA_API_KEY=your_key; kafka-console-producer --broker-list localhost:9092 --producer.config /path/to/config.properties。使用 Spark 内置连接器将 Spark 与 Kafka 连接，并确保集群兼容（例如，Spark 3.x 与 Kafka 2.8+）。对于外部服务，请通过环境变量使用 API 密钥，例如使用 $SPARK_MASTER_URL 连接到 Spark 集群。

错误处理

通过在生产者中实现重试来处理 Kafka 连接错误，例如使用带指数退避的循环：try { producer.send(record) } catch (Exception e) { Thread.sleep(2000 * attempts); }。在 Flink 中，启用重启策略，以便从任务故障中恢复：env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)))。对于 Spark，使用流式查询中的检查点：writeStream.option("checkpointLocation", "/path/to/checkpoints").start()，以便在故障时恢复状态。使用结构化格式记录错误，例如通过 SLF4J，并使用 Prometheus 等工具进行监控，以实现实时告警。

具体用法示例

1. Kafka-Flink 实时日志处理：使用 kafka-console-producer --topic logs --broker-list localhost:9092 将日志导入 Kafka。然后运行 Flink 作业：flink run -c com.example.LogProcessor /path/to/jar --input logs。该作业会筛选错误：env.addSource(new FlinkKafkaConsumer<>("logs", ...)).filter(line -> line.contains("ERROR")).print()。
2. 用于传感器数据聚合的 Spark Streaming：在 Spark 中从 Kafka 读取数据：val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "sensors").load()。聚合数据：df.groupBy(window($"timestamp", "1 minute")).avg("value").writeStream.format("console").start()。这会处理物联网传感器数据流，计算分钟级平均值。

图关系

• 关联集群：data-engineering
• 关联技能：data-ingestion（作为数据源）、machine-learning（用于流上的实时模型推理）
• 依赖项：部署 Kafka/Flink 集群需要容器化等技能