0
0

Flink定制化功能开发，demo代码

前言：

  1.    这是一个Flink自定义开发的基础教学。本文将通过flink的DataStream模块API，以kafka为数据源，构建一个基础测试环境；包含一个kafka生产者线程工具，一个自定义FilterFunction算子，一个自定义MapFunction算子，用一个flink任务的代码逻辑，将实时读kafka并多层处理串起来；让读者体会通过Flink构建自定义函数的技巧。

一、Flink的开发模块分析

Flink提供四个基础模块：核心SDK开发API分别是处理实时计算的DataStream和处理离线计算的DataSet；基于这两个SDK，在其上包装了TableAPI开发模块的SDK；在Table API之上，定义了高度抽象可用SQL开发任务的FlinkSQL。在核心开发API之下，还有基础API的接口，可用于对时间，状态，算子等最细粒度的特性对象做操作，如包装自定义算子的ProcessWindowFunction和ProcessFunction等基础函数以及内置的对象状态StateTtlConfig；

FLINK开发API关系结构如下：

二、定制化开发Demo演示

2.1 场景介绍

Flink实时任务的的通用技术架构是消息队列中间件+Flink任务：

将数据采集到Kafka或pulser这类队列中间件的Topic,然后使用Flink内置的kafkaSource，监控Topic的数据情况，做实时处理。

  1. 这里提供一个kafka的生产者线程，可以自定义构建需要的数据和上传时间，用于控制写入kafka的数据源；
  2. 重写两个DataStream的基础算子：FilterFunction和MapFunction,用于让读者体会，如何对FLINK函数的重新包装，后续更基础的函数原理一样；我这里用String数据对象做处理，减少对象转换的SDK引入，通常要基于业务做数据polo的加工，这个自己处理，将对象换成业务对象；
  3. 然后使用Flink将整个业务串起来，从kafka读数据，经过两层处理，最终输出需要的结果；

2.2 本地demo演示

2.2.1 pom文件

这里以flink1.14.6+scala1.12版本为例：

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.     <parent>
  7.         <groupId>org.example</groupId>
  8.         <artifactId>flinkCDC</artifactId>
  9.         <version>1.0-SNAPSHOT</version>
  10.     </parent>
  12.     <artifactId>flinkStream</artifactId>
  14.     <properties>
  15.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  16.         <maven.compiler.source>8</maven.compiler.source>
  17.         <maven.compiler.target>8</maven.compiler.target>
  18.         <flink-version>1.14.6</flink-version>
  19.         <scala-version>2.12</scala-version>
  20.         <hadop-common-version>2.9.1</hadop-common-version>
  21.         <elasticsearch.version>7.6.2</elasticsearch.version>
  22.         <target.java.version>1.8</target.java.version>
  23.         <scala.binary.version>2.12</scala.binary.version>
  24.         <maven.compiler.source>${target.java.version}</maven.compiler.source>
  25.         <maven.compiler.target>${target.java.version}</maven.compiler.target>
  26.         <log4j.version>2.17.1</log4j.version>
  27.     </properties>
  29.     <repositories>
  30.         <repository>
  31.             <id>apache.snapshots</id>
  32.             <name>Apache Development Snapshot Repository</name>
  33.             <url>https://repository.apache.org/content/repositories/snapshots/</url>
  34.             <releases>
  35.                 <enabled>false</enabled>
  36.             </releases>
  37.             <snapshots>
  38.             </snapshots>
  39.         </repository>
  40.     </repositories>
  41.     <dependencies>
  42.         <dependency>
  43.             <groupId>org.apache.flink</groupId>
  44.             <artifactId>flink-java</artifactId>
  45.             <version>${flink-version}</version>
  46.             <!--            <scope>provided</scope>-->
  47.         </dependency>
  48.         <dependency>
  49.             <groupId>org.apache.flink</groupId>
  50.             <artifactId>flink-core</artifactId>
  51.             <version>${flink-version}</version>
  52.         </dependency>
  53.         <dependency>
  54.             <groupId>org.apache.flink</groupId>
  55.             <artifactId>flink-clients_${scala-version}</artifactId>
  56.             <version>${flink-version}</version>
  57.             <!--            <scope>provided</scope>-->
  58.         </dependency>
  60.         <dependency>
  61.             <groupId>org.apache.flink</groupId>
  62.             <artifactId>flink-connector-kafka_${scala-version}</artifactId>
  63.             <exclusions>
  64.                 <exclusion>
  65.                     <groupId>org.slf4j</groupId>
  66.                     <artifactId>slf4j-log4j12</artifactId>
  67.                 </exclusion>
  68.                 <exclusion>
  69.                     <groupId>org.apache.kafka</groupId>
  70.                     <artifactId>kafka-clients</artifactId>
  71.                 </exclusion>
  72.             </exclusions>
  73.             <version>${flink-version}</version>
  74.         </dependency>
  75.         <dependency>
  76.             <groupId>org.apache.kafka</groupId>
  77.             <artifactId>kafka-clients</artifactId>
  78.             <version>2.4.1</version>
  79.         </dependency>
  81.     </dependencies>
  83.     <build>
  84.         <plugins>
  86.             <!-- Java Compiler -->
  87.             <plugin>
  88.                 <groupId>org.apache.maven.plugins</groupId>
  89.                 <artifactId>maven-compiler-plugin</artifactId>
  90.                 <version>3.1</version>
  91.                 <configuration>
  92.                     <source>${target.java.version}</source>
  93.                     <target>${target.java.version}</target>
  94.                 </configuration>
  95.             </plugin>
  97.             <!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
  98.             <!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
  99.             <plugin>
  100.                 <groupId>org.apache.maven.plugins</groupId>
  101.                 <artifactId>maven-shade-plugin</artifactId>
  102.                 <version>3.1.1</version>
  103.                 <executions>
  104.                     <!-- Run shade goal on package phase -->
  105.                     <execution>
  106.                         <phase>package</phase>
  107.                         <goals>
  108.                             <goal>shade</goal>
  109.                         </goals>
  110.                         <configuration>
  111.                             <createDependencyReducedPom>false</createDependencyReducedPom>
  112.                             <artifactSet>
  113.                                 <excludes>
  114.                                     <exclude>org.apache.flink:flink-shaded-force-shading</exclude>
  115.                                     <exclude>com.google.code.findbugs:jsr305</exclude>
  116.                                     <exclude>org.slf4j:*</exclude>
  117.                                     <exclude>org.apache.logging.log4j:*</exclude>
  118.                                 </excludes>
  119.                             </artifactSet>
  120.                             <filters>
  121.                                 <filter>
  122.                                     <!-- Do not copy the signatures in the META-INF folder.
  123.                                     Otherwise, this might cause SecurityExceptions when using the JAR. -->
  124.                                     <artifact>*:*</artifact>
  125.                                     <excludes>
  126.                                         <exclude>META-INF/*.SF</exclude>
  127.                                         <exclude>META-INF/*.DSA</exclude>
  128.                                         <exclude>META-INF/*.RSA</exclude>
  129.                                     </excludes>
  130.                                 </filter>
  131.                             </filters>
  132.                             <transformers>
  133.                                 <transformer
  134.                                         implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
  135.                                 <transformer
  136.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  137.                                     <mainClass>myflinkml.DataStreamJob</mainClass>
  138.                                 </transformer>
  139.                             </transformers>
  140.                         </configuration>
  141.                     </execution>
  142.                 </executions>
  143.             </plugin>
  144.         </plugins>
  146.         <pluginManagement>
  147.             <plugins>
  149.                 <!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. -->
  150.                 <plugin>
  151.                     <groupId>org.eclipse.m2e</groupId>
  152.                     <artifactId>lifecycle-mapping</artifactId>
  153.                     <version>1.0.0</version>
  154.                     <configuration>
  155.                         <lifecycleMappingMetadata>
  156.                             <pluginExecutions>
  157.                                 <pluginExecution>
  158.                                     <pluginExecutionFilter>
  159.                                         <groupId>org.apache.maven.plugins</groupId>
  160.                                         <artifactId>maven-shade-plugin</artifactId>
  161.                                         <versionRange>[3.1.1,)</versionRange>
  162.                                         <goals>
  163.                                             <goal>shade</goal>
  164.                                         </goals>
  165.                                     </pluginExecutionFilter>
  166.                                     <action>
  167.                                         <ignore/>
  168.                                     </action>
  169.                                 </pluginExecution>
  170.                                 <pluginExecution>
  171.                                     <pluginExecutionFilter>
  172.                                         <groupId>org.apache.maven.plugins</groupId>
  173.                                         <artifactId>maven-compiler-plugin</artifactId>
  174.                                         <versionRange>[3.1,)</versionRange>
  175.                                         <goals>
  176.                                             <goal>testCompile</goal>
  177.                                             <goal>compile</goal>
  178.                                         </goals>
  179.                                     </pluginExecutionFilter>
  180.                                     <action>
  181.                                         <ignore/>
  182.                                     </action>
  183.                                 </pluginExecution>
  184.                             </pluginExecutions>
  185.                         </lifecycleMappingMetadata>
  186.                     </configuration>
  187.                 </plugin>
  188.             </plugins>
  189.         </pluginManagement>
  190.     </build>
  191. </project>
2.2.2 kafka生产者线程方法

  1. package org.example.util;
  3. import org.apache.kafka.clients.producer.KafkaProducer;
  4. import org.apache.kafka.clients.producer.Producer;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import java.util.*;
  9. /**
  10.  * 向kafka生产数据
  11.  *
  12.  * @author i7杨
  13.  * @date 2024/01/12 13:02:29
  14.  */
  16. public class KafkaProducerUtil extends Thread {
  18.     private String topic;
  20.     public KafkaProducerUtil(String topic) {
  21.         super();
  22.         this.topic = topic;
  23.     }
  25.     private static Producer<String, String> createProducer() {
  26.         // 通过Properties类设置Producer的属性
  27.         Properties properties = new Properties();
  28. //        测试环境 kafka 配置
  29.         properties.put("bootstrap.servers", "ip2:9092,ip:9092,ip3:9092");
  30.         properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  31.         properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  32.         return new KafkaProducer<String, String>(properties);
  33.     }
  35.     @Override
  36.     public void run() {
  37.         Producer<String, String> producer = createProducer();
  38.         Random random = new Random();
  39.         Random random2 = new Random();
  41.         while (true) {
  42.             int nums = random.nextInt(10);
  43.             int nums2 = random.nextInt(50);
  44. //            double nums2 = random2.nextDouble();
  46.             String time = new Date().getTime() / 1000 + 5 + "";
  47.             String type = "pv";
  48.             try {
  49.                 if (nums2 % 2 == 0) {
  50.                     type = "pv";
  51.                 } else {
  52.                     type = "uv";
  54.                 }
  55. //                String info = "{\"user\":" + nums + ",\"item\":" + nums * 10 + ",\"category\":" + nums2 + ",\"pv\":" + nums2 * 5 + ",\"ts\":\"" + time + "\"}";
  56.                 String info = nums + "=" + nums2;
  58.                 System.out.println("message : " + info);
  59.                 producer.send(new ProducerRecord<String, String>(this.topic, info));
  60.                 
  61.             } catch (Exception e) {
  62.                 e.printStackTrace();
  63.             }
  64.             System.out.println("=========数据已经写入==========");
  65.             
  66.             try {
  67.                 sleep(1000);
  68.             } catch (InterruptedException e) {
  69.                 e.printStackTrace();
  70.             }
  71.         }
  72.     }
  73.     
  74.     public static void main(String[] args) {
  75.         new KafkaProducerUtil("test01").run();
  76.     }
  77.     
  78.     public static void sendMessage(String topic, String message) {
  79.         Producer<String, String> producer = createProducer();
  80.         producer.send(new ProducerRecord<String, String>(topic, message));
  81.     }
  82.     
  83. }
2.2.3 自定义基础函数

这里自定义了filter和map两个算子函数，测试逻辑按照数据结构变化：

自定义FilterFunction函数算子：阈值小于40的过滤掉

  1. package org.example.funtion;
  3. import org.apache.flink.api.common.functions.FilterFunction;
  5. /**
  6.  * FilterFunction重构
  7.  *
  8.  * @author i7杨
  9.  * @date 2024/01/12 13:02:29
  10.  */
  12. public class InfoFilterFunction implements FilterFunction<String> {
  14.     private double threshold;
  16.     public InfoFilterFunction(double threshold) {
  17.         this.threshold = threshold;
  18.     }
  20.     @Override
  21.     public boolean filter(String value) throws Exception {
  23.         if (value.split("=").length == 2)
  24.             // 阈值过滤
  25.             return Double.valueOf(value.split("=")[1]) > threshold;
  26.         else return false;
  27.     }
  28. }

自定义MapFunction函数：后缀为2的，添加上特殊信息

  1. package org.example.funtion;
  3. import org.apache.flink.api.common.functions.MapFunction;
  5. public class ActionMapFunction implements MapFunction<String, String> {
  7.     @Override
  8.     public String map(String value) throws Exception {
  9.         System.out.println("value:" + value);
  10.         if (value.endsWith("2"))
  11.             return value.concat(":Special processing information");
  12.         else return value;
  13.     }
  14. }
2.2.4 flink任务代码

任务逻辑：使用kafka工具产生数据，然后监控kafka的topic，讲几个函数串起来，输出结果；

  1. package org.example.service;
  3. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  7. import org.apache.kafka.clients.consumer.ConsumerConfig;
  8. import org.apache.kafka.common.serialization.StringDeserializer;
  9. import org.example.funtion.ActionMapFunction;
  10. import org.example.funtion.InfoFilterFunction;
  12. import java.util.*;
  14. public class FlinkTestDemo {
  15.     public static void main(String[] args) throws Exception {
  17.         // 设置执行环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         // Kafka 配置
  21.         Properties kafkaProps = new Properties();
  22.         kafkaProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "ip1:9092,ip2:9092,ip3:9092");
  23.         kafkaProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink-consumer-group");
  24.         kafkaProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  25.         kafkaProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  26.         kafkaProps.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  28.         // 创建 Kafka 消费者
  29.         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
  30.                 "test01",// Kafka 主题名称
  31.                 new SimpleStringSchema(),
  32.                 kafkaProps);
  34.         // 从 Kafka 中读取数据流
  35.         DataStream<String> kafkaStream = env.addSource(kafkaConsumer);
  36.         env.disableOperatorChaining();
  38.         kafkaStream
  39.                 .filter(new InfoFilterFunction(40))
  40.                 .map(new ActionMapFunction())
  41.                 .print("阈值大于40以上的message=");
  43.         // 执行任务
  44.         env.execute("This is a testing task");
  45.     }
  47. }

运行结果：

标签： linq c#
本文转载自: https://blog.csdn.net/weixin_42049123/article/details/135561586
版权归原作者 i7杨 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Flink定制化功能开发，demo代码”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。