#HADOOP_HOME
export HADOOP_HOME=/opt/module/hadoop-3.1.3
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`

./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024 -s 2

./bin/yarn-session.sh -d -n 2 -jm 1024 -tm 1024 -s 2 

./bin/flink run ./examples/batch/WordCount.jar
./bin/flink run ./examples/batch/WordCount.jar --input /opt/tmp/words --output /opt/tmp/test5.txt

yarn application -kill application_1664357090524_0002

./bin/yarn-session.sh 参数

nc -lk 9002

./bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 examples/streaming/SocketWindowWordCount.jar --hostname hadoop102 --port 9002

./bin/flink run -t yarn-per-job  -yjm 1024 -ytm 1024 ./examples/streaming/SocketWindowWordCount.jar --hostname hadoop102 --port 9002

bin/flink run-application -t yarn-application -p 3\
-Dparallelism.default=3 \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dtaskmanager.numberOfTaskSlots=2 \
-Dparallelism.default=10 \
-Dyarn.application.name="application_test" \
./exapmles/batch/WorldCount.jar

yarn application -list

yarn application -kill appID

tar -zxvf flink-1.15.2-bin-scala_2.12.tgz

./bin/start-cluster.sh

./bin/flink run examples/batch/WordCount.jar

Error: VM option ‘UseG1GC’ is experimental and must be enabled via -XX:+UnlockExperimentalVMOptions.
Error: Could not create the Java Virtual Machine.

ID | Hostname      | Name                   | NodeKey | Namespace | IP addresses                  | Ephemeral | Last seen           | Online | Expired
1  | junyao        | junyao-lxgc8rs7        | [kEPjO] | junyao    | 100.64.0.1, fd7a:115c:a1e0::1 | false     | 2022-09-15 02:39:21 | online | no     
2  | homeDS        | homeds-2xqvj919        | [96Nns] | junyao    | 100.64.0.4, fd7a:115c:a1e0::4 | false     | 2022-09-15 02:39:00 | online | no     
3  | VM-0-2-centos | vm-0-2-centos-ieo6arqr | [1M6Gs] | junyao    | 100.64.0.3, fd7a:115c:a1e0::3 | false     | 2022-09-15 02:39:56 | online | no     

- tok: 96.6ms  (Tokyo)
- sfo: 179.8ms (San Francisco)
- sea: 181.3ms (Seattle)
- dfw: 218.6ms (Dallas)
- ord: 229.7ms (Chicago)
- nyc: 233.8ms (New York City)
- fra: 257.7ms (Frankfurt)
- lhr: 259.1ms (London)
- sin: 265ms   (Singapore)
- syd: 317.1ms (Sydney)
- sao: 352.8ms (São Paulo)
- blr: 368.5ms (Bangalore)

version: '3.5'
services:
  derper:
    container_name: derper
    image: fredliang/derper
    restart: always
    volumes:
      - ./cert:/cert
    ports:
      - 3478:3478/udp
      - 23479:23479
    environment:
      DERP_DOMAIN: derp.XXXX.cn
      DERP_ADDR: ":23479"
      DERP_CERT_MODE: manual
      DERP_CERT_DIR: /cert

2022/09/06 01:21:27 no config path specified; using /var/lib/derper/derper.key
2022/09/06 01:21:27 derper: serving on :23479 with TLS
2022/09/06 01:21:27 running STUN server on [::]:3478

# /etc/headscale/config.yaml
derp:
  # List of externally available DERP maps encoded in JSON
  urls:
    - https://controlplane.tailscale.com/derpmap/default

  # Locally available DERP map files encoded in YAML
  #
  # This option is mostly interesting for people hosting
  # their own DERP servers:
  # https://tailscale.com/kb/1118/custom-derp-servers/
  #
  # paths:
  #   - /etc/headscale/derp-example.yaml
  paths:
    - /etc/headscale/derp.yaml

  # If enabled, a worker will be set up to periodically
  # refresh the given sources and update the derpmap
  # will be set up.
  auto_update_enabled: true

  # How often should we check for DERP updates?
  update_frequency: 24h

$ tailscale netcheck
Report:
        * UDP: true
        * IPv4: yes, 14.154.29.252:50713
        * IPv6: yes, [::ffff:14.154.29.252]:50713
        * MappingVariesByDestIP: false
        * HairPinning: false
        * PortMapping: UPnP, NAT-PMP, PCP
        * Nearest DERP: Tencent Guangzhou
        * DERP latency:
                -  gz: 8.5ms   (Tencent Guangzhou)   -----> 这就是我们刚建的derp服务器
                - tok: 96.6ms  (Tokyo)
                - sfo: 179.8ms (San Francisco)
                - sea: 181.3ms (Seattle)
                - dfw: 218.6ms (Dallas)
                - ord: 229.7ms (Chicago)
                - nyc: 233.8ms (New York City)
                - fra: 257.7ms (Frankfurt)
                - lhr: 259.1ms (London)
                - sin: 265ms   (Singapore)
                - syd: 317.1ms (Sydney)
                - sao: 352.8ms (São Paulo)
                - blr: 368.5ms (Bangalore)

tailscale status

100.64.0.1      junyao-lxgc8rs7      junyao       macOS   -
100.64.0.2      homeds-2xqvj919      junyao       linux   active; direct [::ffff:14.154.29.252]:41641, tx 518960 rx 7674032

tailscale ping 100.64.0.2

pong from homeds-2xqvj919 (100.64.0.2) via [::ffff:14.154.29.252]:41641 in 6ms   --->通过点对点
pong from homeds-2xqvj919 (100.64.0.2) via DERP(thk) in 104ms                    --->通过DERP

version: '3.5'
services:
  derper:
    container_name: derper
    image: fredliang/derper
    restart: always
    volumes:
      - ./cert:/cert
      - /var/run/tailscale/tailscaled.sock:/var/run/tailscale/tailscaled.sock
    ports:
      - 3478:3478/udp
      - 23479:23479
    environment:
      DERP_DOMAIN: derp.webhunt.cn
      DERP_ADDR: ":23479"
      DERP_CERT_MODE: manual
      DERP_CERT_DIR: /cert
      DERP_VERIFY_CLIENTS: "true"

version: '3.5'
services:
  headscale:
    image: headscale/headscale:latest-alpine
    container_name: headscale
    volumes:
      - ./config:/etc/headscale
      - ./data/data:/var/lib/headscale
    ports:
      - 8080:8080
    command: headscale serve
    restart: unless-stopped

wget https://github.com/juanfont/headscale/raw/main/config-example.yaml -O config.yaml

headscale namespaces create SOME_NAME

headscale -n NAMESPACE nodes register --key SOME_HEX_VALUE

version: '3.3'
services:
    tailscaled:
        container_name: tailscaled
        image: tailscale/tailscale
        network_mode: host
        privileged: true
        restart: always
        cap_add: 
            - net_admin
            - sys_module
        volumes:
            - ./lib:/var/lib
            - /dev/net/tun:/dev/net/tun
        command: sh -c "mkdir -p /var/run/tailscale && ln -s /tmp/tailscaled.sock /var/run/tailscale/tailscaled.sock && tailscaled"

tailscale up --login-server=http://server_url --accept-routes=true --accept-dns=false

headscale -n default nodes register --key 905cf165204800247fbd33989dbc22be95c987286c45aac3033937041150d846

headscale nodes list

ID | Name | NodeKey | Namespace | IP addresses | Ephemeral | Last seen | Online | Expired
1 | coredns | [Ew3RB] | default | 10.1.0.1 | false | 2022-03-20 09:08:58 | online | no

headscale preauthkeys create -e 24h -n default

headscale -n default preauthkeys list

tailscale up --login-server=http://<HEADSCALE_PUB_IP>:8080 --accept-routes=true --accept-dns=false --authkey $KEY

echo 'net.ipv4.ip_forward = 1' | tee /etc/sysctl.d/ipforwarding.conf
echo 'net.ipv6.conf.all.forwarding = 1' | tee -a /etc/sysctl.d/ipforwarding.conf
sysctl -p /etc/sysctl.d/ipforwarding.conf

tailscale up --login-server=http://<HEADSCALE_PUB_IP>:8080 --accept-routes=true --accept-dns=false --advertise-routes=192.168.100.0/24

headscale nodes list|grep nas
6 | nas | [7LdVc] | default | 10.1.0.6 | false | 2022-03-20 15:50:46 | online | no

headscale routes list -i 6
Route | Enabled

192.168.100.0/24 | false

headscale routes enable -i 6 -r "192.168.100.0/24"
Route | Enabled
192.168.100.0/24 | true

ip route
192.168.100.0/24 dev tailscale0

keyedStream.countWindow(5,2)
  // .max('score')   // 得到的结果中，除了score是符合逻辑的结果外，其他字段是窗口中的第一条的值
  // .min('score')
  // .maxBy('score')  // 得到结果是：最大score所在的那一行数据 
  // .minBy('score')  // 得到结果是：最小score所在的那一行数据
  // .sum('score')    // 得到的结果中，除了score是符合逻辑（score之和）的结果外，其他字段是不可预料的，一直在更新

source.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
  .reduce(new ReduceFunction<long>(){
    @override
    public Long reduce(Long value1,Long value2) throws exception{
      return null
    }
  })

watermarkedBeanStream
  .keyBy(EventBean::getGuid)
  .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
  .aggregate(new AggregateFunction......)

watermarkedBeanStream
  .keyBy(EventBean::getGuid)
  .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
  .appply(new WindowFunction...)

watermarkedBeanStream
  .keyBy(EventBean::getGuid)
  .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
  .process(new WindowFunction...)

package cn.doitedu.flink.java.demos;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2022/5/2
 * @Desc:
 *
 * 测试数据  ：
 * 1,e01,10000,p01,10
 * 1,e02,11000,p02,20
 * 1,e02,12000,p03,40
 * 1,e03,20000,p02,10
 * 1,e01,21000,p03,50
 * 1,e04,22000,p04,10
 * 1,e06,28000,p05,60
 * 1,e07,30000,p02,10
 **/
public class _20_Window_Api_Demo1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        // 1,e01,3000,pg02
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<EventBean2> beanStream = source.map(s -> {
            String[] split = s.split(",");
            return new EventBean2(Long.parseLong(split[0]), split[1], Long.parseLong(split[2]), split[3], Integer.parseInt(split[4]));
        }).returns(EventBean2.class);


        // 分配 watermark ，以推进事件时间
        SingleOutputStreamOperator<EventBean2> watermarkedBeanStream = beanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<EventBean2>forBoundedOutOfOrderness(Duration.ofMillis(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<EventBean2>() {
                            @Override
                            public long extractTimestamp(EventBean2 eventBean, long recordTimestamp) {
                                return eventBean.getTimeStamp();
                            }
                        })
        );

        /**
         * 滚动聚合api使用示例
         * 需求 一 ：  每隔10s，统计最近 30s 的数据中，每个用户的行为事件条数
         * 使用aggregate算子来实现
         */
        SingleOutputStreamOperator<Integer> resultStream = watermarkedBeanStream
                .keyBy(EventBean2::getGuid)
                // 参数1： 窗口长度 ； 参数2：滑动步长
                .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.seconds(10)))
                // reduce ：滚动聚合算子，它有个限制 ，聚合结果的数据类型 与  数据源中的数据类型 ，是一致
                /*.reduce(new ReduceFunction<EventBean>() {
                    @Override
                    public EventBean reduce(EventBean value1, EventBean value2) throws Exception {
                        return null;
                    }
                })*/
                .aggregate(new AggregateFunction<EventBean2, Integer, Integer>() {
                    /**
                     * 初始化累加器
                     * @return
                     */
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    /**
                     * 滚动聚合的逻辑（拿到一条数据，如何去更新累加器）
                     * @param value The value to add
                     * @param accumulator The accumulator to add the value to
                     * @return
                     */
                    @Override
                    public Integer add(EventBean2 value, Integer accumulator) {
                        return accumulator + 1;
                    }

                    /**
                     * 从累加器中，计算出最终要输出的窗口结算结果
                     * @param accumulator The accumulator of the aggregation
                     * @return
                     */
                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }

                    /**
                     *  批计算模式下，可能需要将多个上游的局部聚合累加器，放在下游进行全局聚合
                     *  因为需要对两个累加器进行合并
                     *  这里就是合并的逻辑
                     *  流计算模式下，不用实现！
                     * @param a An accumulator to merge
                     * @param b Another accumulator to merge
                     * @return
                     */
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return a + b;
                    }
                });
        /*resultStream.print();*/


        /**
         * 需求 二 ：  每隔10s，统计最近 30s 的数据中，每个用户的平均每次行为时长
         * 要求用 aggregate 算子来做聚合
         * 滚动聚合api使用示例
         */
        SingleOutputStreamOperator<Double> resultStream2 = watermarkedBeanStream
                .keyBy(EventBean2::getGuid)
                /*.window(TumblingProcessingTimeWindows.of(Time.seconds(30)))*/
                .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.milliseconds(10)))
                // 泛型1： 输入的数据的类型  ；  泛型2： 累加器的数据类型  ；  泛型3： 最终结果的类型
                .aggregate(new AggregateFunction<EventBean2, Tuple2<Integer, Integer>, Double>() {
                    @Override
                    public Tuple2<Integer, Integer> createAccumulator() {
                        return Tuple2.of(0, 0);
                    }

                    @Override
                    public Tuple2<Integer, Integer> add(EventBean2 eventBean, Tuple2<Integer, Integer> accumulator) {
                        // accumulator.setField(accumulator.f0+1,0);
                        // accumulator.setField(accumulator.f1+eventBean.getActTimelong(),1);
                        // return accumulator;

                        return Tuple2.of(accumulator.f0 + 1, accumulator.f1 + eventBean.getActTimelong());
                    }

                    @Override
                    public Double getResult(Tuple2<Integer, Integer> accumulator) {

                        return accumulator.f1 / (double) accumulator.f0;
                    }

                    /**
                     * 在批计算模式中，shuffle的上游可以做局部聚合，然后会把局部聚合结果交给下游去做全局聚合
                     * 因此，就需要提供  两个局部聚合结果进行合并的逻辑
                     *
                     * 在流式计算中，不存在这种 上游局部聚合和交给下游全局聚合的机制！
                     * 所以，在流式计算模式下，不用实现下面的方法
                     * @param a An accumulator to merge
                     * @param b Another accumulator to merge
                     * @return
                     */
                    @Override
                    public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
                        return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
                    }
                });
        /*resultStream2.print();*/


        /**
         * TODO 补充练习 1
         * 需求 一 ：  每隔10s，统计最近 30s 的数据中，每个用户的行为事件条数
         * 滚动聚合api使用示例
         * 使用sum算子来实现
         */
        watermarkedBeanStream
                .map(bean->Tuple2.of(bean,1)).returns(new TypeHint<Tuple2<EventBean2, Integer>>() {})
                .keyBy(tp->tp.f0.getGuid())
                .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
                // 数据: Tuple2<Bean,1>
                .sum("f1")
                /*.print()*/;


        /**
         * TODO 补充练习 2
         * 需求 一 ：  每隔10s，统计最近 30s 的数据中，每个用户的最大行为时长
         * 滚动聚合api使用示例
         * 用max算子来实现
         */
        watermarkedBeanStream
                .keyBy(EventBean2::getGuid)
                .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
                .max("actTimelong")
                /*.print()*/;



        /**
         * TODO 补充练习 3
         * 需求 一 ：  每隔10s，统计最近 30s 的数据中，每个用户的最大行为时长及其所在的那条行为记录
         * 滚动聚合api使用示例
         * 用maxBy算子来实现
         */
        watermarkedBeanStream
                .keyBy(EventBean2::getGuid)
                .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
                .maxBy("actTimelong")
                /*.print()*/;



        /**
         * TODO 补充练习 4
         * 需求 一 ：  每隔10s，统计最近 30s 的数据中，每个页面上发生的行为中，平均时长最大的前2种事件及其平均时长
         * 用 process算子来实现
         */
        watermarkedBeanStream
                .keyBy(bean->bean.getPageId())
                .window(SlidingEventTimeWindows.of(Time.seconds(30),Time.seconds(10)))
                .process(new ProcessWindowFunction<EventBean2, Tuple3<String,String,Double>, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<EventBean2, Tuple3<String,String,Double>, String, TimeWindow>.Context context, Iterable<EventBean2> elements, Collector<Tuple3<String,String,Double>> out) throws Exception {
                        // 构造一个hashmap来记录每一个事件的发生总次数，和行为总时长
                        HashMap<String, Tuple2<Integer, Long>> tmpMap = new HashMap<>();

                        // 遍历窗口中的每一条数据
                        for (EventBean2 element : elements) {
                            String eventId = element.getEventId();
                            Tuple2<Integer, Long> countAndTimelong = tmpMap.getOrDefault(eventId,Tuple2.of(0,0L));

                            tmpMap.put(eventId,Tuple2.of(countAndTimelong.f0+1,countAndTimelong.f1+element.getActTimelong()) );
                        }

                        // 然后，从tmpMap中，取到 平均时长 最大的前两个事件
                        ArrayList<Tuple2<String, Double>> tmpList = new ArrayList<>();
                        for (Map.Entry<String, Tuple2<Integer, Long>> entry : tmpMap.entrySet()) {
                            String eventId = entry.getKey();
                            Tuple2<Integer, Long> tuple = entry.getValue();
                            double avgTimelong = tuple.f1/ (double)tuple.f0;
                            tmpList.add(Tuple2.of(eventId,avgTimelong));
                        }

                        // 然后对tmpList按平均时长排序
                        Collections.sort(tmpList, new Comparator<Tuple2<String, Double>>() {
                            @Override
                            public int compare(Tuple2<String, Double> tp1, Tuple2<String, Double> tp2) {
                               /* return tp2.f1.compareTo(tp1.f1);*/
                                return Double.compare(tp2.f1,tp1.f1);
                            }
                        });

                        // 输出前2个
                        for(int i=0;i<Math.min(tmpList.size(),2);i++){
                            out.collect(Tuple3.of(key,tmpList.get(i).f0,tmpList.get(i).f1));
                        }
                    }
                })
                .print();



        /**
         * 全窗口计算api使用示例
         * 需求 三 ：  每隔10s，统计最近 30s 的数据中，每个用户的行为事件中，行为时长最长的前2条记录
         * 要求用 apply 或者  process 算子来实现
         *
         */
        // 1. 用apply算子来实现需求
        SingleOutputStreamOperator<EventBean2> resultStream3 = watermarkedBeanStream.keyBy(EventBean2::getGuid)
                .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.seconds(10)))

                // 泛型1： 输入数据类型；  泛型2：输出结果类型；  泛型3： key的类型， 泛型4：窗口类型
                .apply(new WindowFunction<EventBean2, EventBean2, Long, TimeWindow>() {
                    /**
                     *
                     * @param key 本次传给咱们的窗口是属于哪个key的
                     * @param window 本次传给咱们的窗口的各种元信息（比如本窗口的起始时间，结束时间）
                     * @param input 本次传给咱们的窗口中所有数据的迭代器
                     * @param out 结果数据输出器
                     * @throws Exception
                     */
                    @Override
                    public void apply(Long key, TimeWindow window, Iterable<EventBean2> input, Collector<EventBean2> out) throws Exception {

                        // low bi写法： 从迭代器中迭代出数据，放入一个arraylist，然后排序，输出前2条
                        ArrayList<EventBean2> tmpList = new ArrayList<>();

                        // 迭代数据，存入list
                        for (EventBean2 eventBean2 : input) {
                            tmpList.add(eventBean2);
                        }
                        // 排序
                        Collections.sort(tmpList, new Comparator<EventBean2>() {
                            @Override
                            public int compare(EventBean2 o1, EventBean2 o2) {
                                return o2.getActTimelong() - o1.getActTimelong();
                            }
                        });

                        // 输出前2条
                        for (int i = 0; i < Math.min(tmpList.size(), 2); i++) {
                            out.collect(tmpList.get(i));
                        }

                    }
                });
        /*resultStream3.print();*/


        // 2. 用process算子来实现需求
        SingleOutputStreamOperator<String> resultStream4 = watermarkedBeanStream.keyBy(EventBean2::getGuid)
                .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.seconds(10)))
                .process(new ProcessWindowFunction<EventBean2, String, Long, TimeWindow>() {
                    @Override
                    public void process(Long aLong, ProcessWindowFunction<EventBean2, String, Long, TimeWindow>.Context context, Iterable<EventBean2> input, Collector<String> out) throws Exception {

                        // 本次窗口的元信息
                        TimeWindow window = context.window();
                        long maxTimestamp = window.maxTimestamp();// 本窗口允许的最大时间戳  [1000,2000) ，其中 1999就是允许的最大时间戳； 2000就是窗口的end
                        long windowStart = window.getStart();
                        long windowEnd = window.getEnd();


                        // low bi写法： 从迭代器中迭代出数据，放入一个arraylist，然后排序，输出前2条
                        ArrayList<EventBean2> tmpList = new ArrayList<>();

                        // 迭代数据，存入list
                        for (EventBean2 eventBean2 : input) {
                            tmpList.add(eventBean2);
                        }
                        // 排序
                        Collections.sort(tmpList, new Comparator<EventBean2>() {
                            @Override
                            public int compare(EventBean2 o1, EventBean2 o2) {
                                return o2.getActTimelong() - o1.getActTimelong();
                            }
                        });

                        // 输出前2条
                        for (int i = 0; i < Math.min(tmpList.size(), 2); i++) {
                            EventBean2 bean = tmpList.get(i);
                            out.collect( "窗口start:"+windowStart + "," +"窗口end:"+ windowEnd + "," +  bean.getGuid() + "," + bean.getEventId() + "," + bean.getTimeStamp() + "," +bean.getPageId() + "," +bean.getActTimelong());
                        }
                    }

                });
        /*resultStream4.print();*/
        env.execute();
    }
}

version: '2.1'
services:
   sqlserver:
     image: mcr.microsoft.com/mssql/server:2019-latest
     container_name: sqlserver
     ports:
       - "1433:1433"
     environment:
       - "MSSQL_AGENT_ENABLED=true"
       - "MSSQL_PID=Standard"
       - "ACCEPT_EULA=Y"
       - "SA_PASSWORD=Password!"

-- Sqlserver
CREATE DATABASE inventory;
GO
USE inventory;
EXEC sys.sp_cdc_enable_db;

-- Create and populate our products using a single insert with many rows
CREATE TABLE products (
id INTEGER IDENTITY(101,1) NOT NULL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description VARCHAR(512),
weight FLOAT
);
INSERT INTO products(name,description,weight)
VALUES ('scooter','Small 2-wheel scooter',3.14);
INSERT INTO products(name,description,weight)
VALUES ('car battery','12V car battery',8.1);
INSERT INTO products(name,description,weight)
VALUES ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8);
INSERT INTO products(name,description,weight)
VALUES ('hammer','12oz carpenter''s hammer',0.75);
INSERT INTO products(name,description,weight)
VALUES ('hammer','14oz carpenter''s hammer',0.875);
INSERT INTO products(name,description,weight)
VALUES ('hammer','16oz carpenter''s hammer',1.0);
INSERT INTO products(name,description,weight)
VALUES ('rocks','box of assorted rocks',5.3);
INSERT INTO products(name,description,weight)
VALUES ('jacket','water resistent black wind breaker',0.1);
INSERT INTO products(name,description,weight)
VALUES ('spare tire','24 inch spare tire',22.2);
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'products', @role_name = NULL, @supports_net_changes = 0;
-- Create some very simple orders
CREATE TABLE orders (
id INTEGER IDENTITY(10001,1) NOT NULL PRIMARY KEY,
order_date DATE NOT NULL,
purchaser INTEGER NOT NULL,
quantity INTEGER NOT NULL,
product_id INTEGER NOT NULL,
FOREIGN KEY (product_id) REFERENCES products(id)
);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('16-JAN-2016', 1001, 1, 102);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('17-JAN-2016', 1002, 2, 105);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('19-FEB-2016', 1002, 2, 106);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('21-FEB-2016', 1003, 1, 107);
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'orders', @role_name = NULL, @supports_net_changes = 0;
GO

CREATE TABLE products (
  id INT,
  name STRING,
  description STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname' = '192.168.31.6',
  'port' = '1433',
  'username' = 'sa',
  'password' = 'Password!',
  'database-name' = 'inventory',
  'schema-name' = 'dbo',
  'table-name' = 'products'
);


CREATE TABLE orders (
  id INT,
  order_date DATE,
  purchaser INT,
  quantity INT,
  product_id INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname' = '192.168.31.6',
  'port' = '1433',
  'username' = 'sa',
  'password' = 'Password!',
  'database-name' = 'inventory',
  'schema-name' = 'dbo',
  'table-name' = 'orders'
);


CREATE TABLE enriched_orders (
  order_id INT,
  order_date DATE,
  purchaser INT,
  quantity INT,
  product_name STRING,
  product_description STRING
) WITH ('connector' = 'print');


INSERT INTO
  enriched_orders
SELECT
  o.id,
  o.order_date,
  o.purchaser,
  o.quantity,
  p.name,
  p.description
FROM
  orders AS o
  LEFT JOIN products AS p ON o.product_id = p.id;

{
    "_id": ObjectId("5f601cd6afef2b4993dc7afd"),
    "created": ISODate("2020-09-15T01:45:44.753Z"),
    "mfrId": ObjectId("5ea279be76860394d14a4982"),
    "mfrName": "Biolegend",
    "name": "FITC anti-mouse CD2",
    "ras": "100105-BLG",
    "sn": "100105",
    "spec": "50 μg",
    "status": true,
    "taxrate": 13,
    "unit": "EA",
    "updated": ISODate("2021-01-15T08:55:45.668Z"),
    "price": NumberInt("980"),
    "taxcode": "107030710",
    "clone": "RM2-5",
    "lastOrderAt": ISODate("2021-08-05T03:09:47.577Z"),
    "manual": "https://www.biolegend.com/Default.aspx?ID=6664&productid=472",
    "pn": "472",
    "cumulativeSales": NumberInt("0"),
    "isDeprecated": false,
    "ship": "蓝冰",
    "storage": "2°C-8°C",
    "isPublic": true,
    "invtCode": "1405.01"
}

CREATE TABLE material (
  _id STRING,
  created TIMESTAMP_LTZ(3),
  mfrId STRING,
  mfrName STRING,
  name STRING,
  ras STRING,
  sn STRING,
  spec STRING,
  status BOOLEAN,
  taxrate INT,
  unit STRING,
  updated TIMESTAMP_LTZ(3),
  price DECIMAL(10, 5),
  taxcode STRING,
  clone STRING,
  lastOrderAt TIMESTAMP_LTZ(3),
  manual STRING,
  pn STRING,
  cumulativeSales INT,
  isDeprecated BOOLEAN,
  ship STRING,
  storage STRING,
  isPublic BOOLEAN,
  invtCode STRING,
  PRIMARY KEY (_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb-cdc',
  'hosts' = 'localhost:27017',
  'username' = 'XXX',
  'password' = 'XXX',
  'database' = 'biocitydb',
  'collection' = 'material'
);
CREATE TABLE es_material (
  _id STRING,
  created TIMESTAMP_LTZ(3),
  mfrId STRING,
  mfrName STRING,
  name STRING,
  ras STRING,
  sn STRING,
  spec STRING,
  status BOOLEAN,
  taxrate INT,
  unit STRING,
  updated TIMESTAMP_LTZ(3),
  price DECIMAL(10, 5),
  taxcode STRING,
  clone STRING,
  lastOrderAt TIMESTAMP_LTZ(3),
  manual STRING,
  pn STRING,
  cumulativeSales INT,
  isDeprecated BOOLEAN,
  ship STRING,
  storage STRING,
  isPublic BOOLEAN,
  invtCode STRING,
  PRIMARY KEY (_id) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch-7',
  'hosts' = 'http://localhost:9200',
  'index' = 'es_material'
);
INSERT INTO
  es_material
select
  _id,
  created,
  mfrId,
  mfrName,
  name,
  ras,
  sn,
  spec,
  status,
  taxrate,
  unit,
  updated,
  price,
  taxcode,
  clone,
  lastOrderAt,
  manual,
  pn,
  cumulativeSales,
  isDeprecated,
  ship,
  storage,
  isPublic,
  invtCode
from
  material;

<dependency>
  <groupId>com.ververica</groupId>
  <artifactId>flink-connector-mongodb-cdc</artifactId>
  <version>2.2.1</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
  <version>1.14.4</version>
</dependency>