Flink核心概念

public class DistributedCacheDemo {
?
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //1：注册缓存文件,可以读取hdfs上的文件,也可以本地文件
        //nameAgeConfig.txt文件内容：
            //jack,22
            //lily,20
            //zhangsan,21
            //lisi,25
            //wanger,26
        env.registerCachedFile("C:\\Users\\ksw\\Desktop\\nameAgeConfig.txt", "nameAgeConfig");
?
        // 生成测试数据
        DataSource<String> data = env.fromElements("jack", "zhangsan");
?
        // 从分布式缓存中获取每个人对应的年龄
        MapOperator<String, String> result = data.map(new RichMapFunction<String, String>() {
            private Map<String, String> nameAgeMap = new HashMap<>();
?
            //重写open方法
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //2：可根据注册的名字使用该缓存文件
                File cacheFile = getRuntimeContext().getDistributedCache().getFile("nameAgeConfig");
                List<String> lines = FileUtils.readLines(cacheFile);
                for (String line : lines) {
                    String[] split = line.split(",");
                    this.nameAgeMap.put(split[0], split[1]);
                }
            }
?
            @Override
            public String map(String name) throws Exception {
                //处理逻辑
                String age = nameAgeMap.get(name);
                return name + ":" + age;
            }
        });
?
        result.printToErr();
    }
}

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
?
DataStream<String> text = [...]
DataStream<Tuple2<String, Integer>> wordCounts = text
    .flatMap(new LineSplitter())
    .keyBy(0)
    .timeWindow(Time.seconds(5))
    .sum(1).setParallelism(5);
?
wordCounts.print();
?
env.execute("Word Count Example");

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);
?
DataStream<String> text = [...]
DataStream<Tuple2<String, Integer>> wordCounts = [...]
wordCounts.print();
?
env.execute("Word Count Example");

./bin/flink run -p 10 ../examples/*WordCount-java*.jar

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
?
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
?
// 另外也可以设置时间类型为IngestionTime或EventTime:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime); 
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
?
DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));
?
stream
    .keyBy( (event) -> event.getUser() )
    .timeWindow(Time.hours(1))
    .reduce( (a, b) -> a.add(b) )
    .addSink(...);

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
?
DataStream<MyEvent> stream = env.readFile(
        myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
        FilePathFilter.createDefaultFilter(), typeInfo);
?
// 在这里调用了assignTimestampsAndWatermarks()方法，并且传入我们自己定义的new MyTimestampsAndWatermarks()
DataStream<MyEvent> withTimestampsAndWatermarks = stream
        .filter( event -> event.severity() == WARNING )
        .assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks());
?
withTimestampsAndWatermarks
        .keyBy( (event) -> event.getGroup() )
        .timeWindow(Time.seconds(10))
        .reduce( (a, b) -> a.add(b) )
        .addSink(...);

@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
    this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
    if (characteristic == TimeCharacteristic.ProcessingTime) {
        getConfig().setAutoWatermarkInterval(0);
    } else {
        getConfig().setAutoWatermarkInterval(200);
    }
}

public static void main(String[] args) throws Exception {
?
    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
?
    //设置为eventtime事件类型
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    //设置水印生成时间间隔100ms
    env.getConfig().setAutoWatermarkInterval(100);
?
    DataStream<String> dataStream = env
            .socketTextStream("127.0.0.1", 9000)
            .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
                private Long currentTimeStamp = 0L;
                //maxOutOfOrderness参数指的是允许数据乱序的时间范围。简单说，允许数据迟到maxOutOfOrderness 这么长的时间
                private Long maxOutOfOrderness = 5000L;
?
                @Override
                public Watermark getCurrentWatermark() {
?
                    return new Watermark(currentTimeStamp - maxOutOfOrderness);
                }
?
                @Override
                public long extractTimestamp(String s, long l) {
                    String[] arr = s.split(",");
                    long timeStamp = Long.parseLong(arr[1]);
                    currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
                    System.err.println(s + ",EventTime:" + timeStamp + ",watermark:" + (currentTimeStamp - maxOutOfOrderness));
                    return timeStamp;
                }
            });
?
    dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {
        @Override
        public Tuple2<String, Long> map(String s) throws Exception {
?
            String[] split = s.split(",");
            return new Tuple2<String, Long>(split[0], Long.parseLong(split[1]));
        }
    })
            .keyBy(0)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .minBy(1)
            .print();
?
    env.execute("WaterMark Test Demo");
?
}

data.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<UserActionRecord>() {
?
      @Override
      public Watermark checkAndGetNextWatermark(MyData data, long l) {
        return data.getRecord.startsWith("watermark") ? new Watermark(l) : null;
      }
?
      @Override
      public long extractTimestamp(MyData data, long l) {
        return data.getTimestamp();
      }
    });
    
class MyData{
    private String record;
    private Long timestamp;
    public String getRecord() {
        return record;
    }
    public void setRecord(String record) {
        this.record = record;
    }
    public Timestamp getTimestamp() {
        return timestamp;
    }
    public void setTimestamp(Timestamp timestamp) {
        this.timestamp = timestamp;
    }
}