第一个Flink例子-flink写入kafka

迷南。 2022-10-24 13:16 340阅读 0赞

第一个Flink例子-flink写入kafka

  • 依赖
  • 代码
  • 运行

依赖

1.zookeeper

单机模式，参考：https://blog.csdn.net/sndayYU/article/details/90718238

基本上绿色版改下conf/zoo.cfg的日志和数据目录、建立下对应目录。

2.kafka

单机模式，参考：https://blog.csdn.net/sndayYU/article/details/90718786

基本上建立下 D:\tmp\kafka-logs 这个目录就ok了

代码

1.pom.xml

  1. <properties>
  2.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3.         <maven.compiler.source>1.8</maven.compiler.source>
  4.         <maven.compiler.target>1.8</maven.compiler.target>
  5.     </properties>
  7.     <dependencies>
  8.         <dependency>
  9.             <groupId>junit</groupId>
  10.             <artifactId>junit</artifactId>
  11.             <version>4.13</version>
  12.             <scope>test</scope>
  13.         </dependency>
  14.         <dependency>
  15.             <groupId>org.apache.flink</groupId>
  16.             <artifactId>flink-connector-kafka_2.12</artifactId>
  17.             <version>1.8.0</version>
  18.         </dependency>
  20.         <dependency>
  21.             <groupId>org.apache.flink</groupId>
  22.             <artifactId>flink-java</artifactId>
  23.             <version>1.8.0</version>
  24.             <scope>provided</scope>
  25.         </dependency>
  26.         <dependency>
  27.             <groupId>org.apache.flink</groupId>
  28.             <artifactId>flink-streaming-java_2.11</artifactId>
  29.             <version>1.8.0</version>
  30.             <scope>provided</scope>
  31.         </dependency><dependency>
  32.         <groupId>org.apache.flink</groupId>
  33.             <artifactId>flink-java</artifactId>
  34.             <version>1.8.0</version>
  35.             <scope>compile</scope>
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.flink</groupId>
  39.             <artifactId>flink-streaming-java_2.11</artifactId>
  40.             <version>1.8.0</version>
  41.             <scope>compile</scope>
  42.         </dependency>
  43.     </dependencies>

2.FlinkToKafka

  1. package com.ydfind;
  3. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  4. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  7. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
  9. import java.text.SimpleDateFormat;
  10. import java.util.*;
  12. /**
  13.  * flink不断产生数据 发送到 kafka
  14.  * @author drd
  15.  * @date 2021.1.29
  16.  */
  17. public class FlinkToKafka {
  18.     public static void main(String[] args) throws Exception{
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         // 注册数据源
  22.         DataStreamSource<String> text = env.addSource(new MyFlinkProducer()).setParallelism(1);
  24.         Properties properties = new Properties();
  25.         properties.setProperty("bootstrap.servers", "localhost:9092");
  26.         FlinkKafkaProducer<String> producer = new FlinkKafkaProducer("flink-test",new SimpleStringSchema(),properties);
  27.         producer.setWriteTimestampToKafka(true);
  29.         // 添加处理对象
  30.         text.addSink(producer);
  32.         // 开始执行
  33.         env.execute();
  34.     }
  36.     public static class MyFlinkProducer implements SourceFunction<String> {//1
  38.         private boolean running = true;
  40.         @Override
  41.         public void run(SourceContext<String> sourceContext) throws Exception {
  42.             String format = "yyyy-MM-dd HH:mm:ss.SSS";
  43.             SimpleDateFormat sdf = new SimpleDateFormat(format);
  44.             while(running){
  45.                 String msg = sdf.format(new Date(System.currentTimeMillis())) + ", this is from MySourceProducer";
  46.                 sourceContext.collect(msg);
  47.                 Thread.sleep(3000);
  48.             }
  49.         }
  51.         @Override
  52.         public void cancel() {
  53.             running = false;
  54.         }
  55.     }
  56. }

运行

1.运行zookeeper

  1. win10:
  2. .\bin\zkServer.cmd
  4. contenos:
  5. .\bin\zkServer.sh start

2.kafka

  1. // 创建kafka的topic
  2. cd D:\YDGreenNew\kafka_2.12-2.2.0
  3. bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic flink-test

3.用idea运行代码

4.观察kafka的消费者输出

  1. // 启动消费者观察结果
  2. .\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic flink-test --from-beginning

在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，340人围观）

还没有评论，来说两句吧...

相关阅读