Flink自定义事件事件和水位线

刺骨的言语ヽ痛彻心扉 2022-10-16 03:56 157阅读 0赞

**目录**

主类

pom.xml

结果输出

--------------------

## 主类 ##

package myflink.EventTime;
    
    import org.apache.flink.api.common.eventtime.*;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.ProcessFunction;
    import org.apache.flink.util.Collector;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    
    /**
     * @author pangsl
     * @date 2021/5/25 17:02
     * @Description Flink自定义事件事件和水位线
     */
    public class WindowAndWatermarkDemo {
        public static void main(String[] args) throws Exception {
    
    
            StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置使用EventTime，默认使用processtime
            sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            //设置并行度为1,默认并行度是当前机器的CPU数量
            sEnv.setParallelism(1);
    
    
            DataStream<String> input = sEnv.addSource(new MySource());
            /*消息格式：String,time。例如：消息1,15115556459354*/
            //解析输入的数据
            DataStream<Tuple2<String, Long>> inputMap = input.map(new MapFunction<String, Tuple2<String, Long>>() {
    
                @Override
                public Tuple2<String, Long> map(String value) throws Exception {
                    String[] arr = value.split(",");
                    return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
                }
            });
            //抽取timestamp，生成Watermark，这里使用的是Flink1.11最新版的水印生成策略
            DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {
    
                @Override
                public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                    return new WatermarkGenerator<Tuple2<String, Long>>() {
                        private long maxTimestamp;
                        private long delay = 3000;
    
                        @Override
                        public void onEvent(Tuple2<String, Long> event, long eventTimestamp, WatermarkOutput output) {
                            maxTimestamp = Math.max(maxTimestamp, event.f1);
    
                        }
    
                        @Override
                        public void onPeriodicEmit(WatermarkOutput output) {
                            output.emitWatermark(new Watermark(maxTimestamp - delay));
                        }
                    };
                }
            });
            /*获取水印信息*/
            waterMarkStream.process(new ProcessFunction<Tuple2<String, Long>, Object>() {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    
                @Override
                public void processElement(Tuple2<String, Long> value, Context ctx, Collector<Object> out) throws Exception {
                    long w = ctx.timerService().currentWatermark();
                    System.out.println(" 水印 ： " + w + "水印时间" + sdf.format(w) + "消息的事件时间" +
                            sdf.format(value.f1));
                }
            });
            waterMarkStream.print();
            sEnv.execute();
        }
    }

## pom.xml ##

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <parent>
            <artifactId>flinkdemo</artifactId>
            <groupId>com.flink.pangsl</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>Flink_shizhanpai</artifactId>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    
        <properties>
            <flink.version>1.11.0</flink.version>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        </properties>
    
        <dependencies>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner-blink_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-common</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
    
        </dependencies>
    
    </project>

## 结果输出 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70]: /images/20221014/537d399b87904ee1a880385ae0751803.png