Flink从Kafka读取数据的完整指南

目录

代码解析

方法解析

代码解释

进阶配置

完整代码示例

运行步骤

常见问题

---

代码解析

1. 导入依赖

   • org.apache.flink.streaming.api.scala._：Flink 的 Scala API。
   • org.apache.flink.connector.kafka.source.KafkaSource：Flink 提供的 Kafka Source 连接器。
   • org.apache.flink.api.common.serialization.SimpleStringSchema：用于将 Kafka 消息反序列化为字符串。

2. 主程序

   • StreamExecutionEnvironment.getExecutionEnvironment：获取 Flink 的执行环境。
   • KafkaSource.builder()：构建 Kafka Source。
     • setBootstrapServers("")：设置 Kafka 的 broker 地址（需要填写实际的 Kafka 地址，如 localhost:9092）。
     • setTopics("")：设置要消费的 Kafka topic（需要填写实际的 topic 名称）。
     • setValueOnlyDeserializer(new SimpleStringSchema())：设置消息的反序列化器，将 Kafka 消息反序列化为字符串。
   • env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source")：将 Kafka Source 转换为 Flink 的 DataStream。
   • kafkaStream.print("kafka")：将数据流打印到控制台。
   • env.execute("SourceKafka")：启动 Flink 作业。

package sourceimport java.util.Propertiesimport org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.streaming.api.scala._/**** @PROJECT_NAME: flink1.13* @PACKAGE_NAME: source* @author: 赵嘉盟-HONOR* @data: 2023-11-19 1:54* @DESCRIPTION**/
object SourceKafka {def main(args: Array[String]): Unit = {val env=StreamExecutionEnvironment.getExecutionEnvironmentval kafkaSource =KafkaSource.builder().setBootstrapServers("").setTopics("").setValueOnlyDeserializer(new SimpleStringSchema()).build()val kafkaStream: DataStream[String] =env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source")kafkaStream.print("kafka")env.execute("SourceKafka")}
}

方法解析

1. val env=StreamExecutionEnvironment.getExecutionEnvironment: 创建一个Flink流处理执行环境，并将其赋值给变量env。
2. val kafkaSource = KafkaSource.builder(): 创建一个Kafka源构建器对象。
3. .setBootstrapServers(""): 设置Kafka集群的地址，这里为空字符串表示使用默认的Kafka集群地址。
4. .setTopics(""): 设置要消费的Kafka主题，这里为空字符串表示消费所有主题。
5. .setValueOnlyDeserializer(new SimpleStringSchema()): 设置值的反序列化器，这里使用SimpleStringSchema将消息的值解析为字符串类型。
6. .build(): 构建Kafka源对象。
7. val kafkaStream: DataStream[String] = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source"): 使用Flink流处理执行环境env从Kafka源kafkaSource中创建数据流，并指定没有水印策略和数据流名称为"kafka Source"。
8. kafkaStream.print("kafka"): 打印数据流的内容，输出到控制台，并指定输出的前缀为"kafka"。

代码解释

1. StreamExecutionEnvironment:

   • val env = StreamExecutionEnvironment.getExecutionEnvironment：创建 Flink 流处理环境。

2. KafkaSource:

   • KafkaSource.builder()：构建 Kafka 数据源。
   • .setBootstrapServers("")：设置 Kafka 集群的地址（例如：localhost:9092）。
   • .setTopics("")：设置要消费的 Kafka 主题（例如：my-topic）。
   • .setValueOnlyDeserializer(new SimpleStringSchema())：设置消息的反序列化器，这里使用 SimpleStringSchema 将消息反序列化为字符串。

3. env.fromSource:

   • 将 Kafka 数据源添加到 Flink 流处理环境中。
   • WatermarkStrategy.noWatermarks()：不生成水印（适用于无事件时间处理的场景）。
   • "kafka Source"：为数据源指定一个名称。

4. kafkaStream.print("kafka"):

   • 打印从 Kafka 消费的消息，每条消息前加上 "kafka" 前缀。

5. env.execute("SourceKafka"):

   • 启动 Flink 作业，作业名称为 "SourceKafka"。

进阶配置

1. 添加 Kafka 配置

   • 如果需要额外的 Kafka 配置（如消费者组 ID），可以通过 setProperties 方法设置：

     val props = new Properties()
     props.setProperty("group.id", "flink-consumer-group")
     kafkaSource.setProperties(props)

1. Watermark 策略

   • 如果你的数据流需要处理事件时间，可以自定义 WatermarkStrategy，而不是使用 WatermarkStrategy.noWatermarks()。例如：

     scala

     WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10))

完整代码示例

以下是改进后的代码：

package sourceimport java.util.Properties
import java.time.Durationimport org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.streaming.api.scala._object SourceKafka {def main(args: Array[String]): Unit = {// 获取 Flink 执行环境val env = StreamExecutionEnvironment.getExecutionEnvironment// Kafka 配置val props = new Properties()props.setProperty("group.id", "flink-consumer-group")// 构建 Kafka Sourceval kafkaSource = KafkaSource.builder().setBootstrapServers("localhost:9092") // Kafka broker 地址.setTopics("test-topic") // Kafka topic.setValueOnlyDeserializer(new SimpleStringSchema()) // 反序列化器.setProperties(props) // Kafka 消费者配置.build()// 从 Kafka Source 创建 DataStreamval kafkaStream: DataStream[String] = env.fromSource(kafkaSource,WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10)), // Watermark 策略"kafka Source")// 打印数据流kafkaStream.print("kafka")// 启动 Flink 作业env.execute("SourceKafka")}
}

---

运行步骤

1. 启动 Kafka：确保 Kafka 服务已启动，并创建了指定的 topic（如 test-topic）。
2. 发送测试数据：使用 Kafka 生产者向 test-topic 发送一些测试消息。
3. 运行程序：运行 Flink 程序，观察控制台输出。

---

常见问题

1. Kafka 连接失败
   • 检查 Kafka 服务是否正常运行，以及 bootstrap.servers 配置是否正确。
2. No Watermark Strategy
   • 如果不需要事件时间处理，可以继续使用 WatermarkStrategy.noWatermarks()。
3. 依赖冲突
   • 确保 Flink 和 Kafka 的版本兼容。