DataStreamSource<String> source = env.readTextFile("/tmp/data");

DataStreamSource<String> source = env.socketTextStream("127.0.0.1", 8089);

DataStreamSource<String> source = env.fromCollection(collection);


DataStream<SensorReading> source = env.fromElements(new Object())

<flink.version>1.13.2</flink.version>


<dependency>  
    <groupId>org.apache.flink</groupId>  
    <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>  
    <version>${flink.version}</version>  
    <scope>provided</scope>  
</dependency>  
<dependency>  
    <groupId>org.apache.flink</groupId>  
    <artifactId>flink-connector-kafka_2.12</artifactId>  
    <version>${flink.version}</version>  
</dependency>  
  
<dependency>  
    <groupId>org.apache.flink</groupId>  
    <artifactId>flink-connector-base</artifactId>  
    <version>${flink.version}</version>  
</dependency>

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()  
        .setBootstrapServers("192.168.1.100:9092")  
        .setTopics("test_topic")  
        .setGroupId("flink-demo")  
        .setStartingOffsets(OffsetsInitializer.earliest())  
        .setValueOnlyDeserializer(new SimpleStringSchema())  
        .build();  
  
DataStreamSource<String> source = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");

Properties consumerProperties = new Properties();  
consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.100:9092");  
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "flink-demo");  
DataStreamSource<String> source =  
        env.addSource(new FlinkKafkaConsumer<>("test_topic", new SimpleStringSchema(), consumerProperties));

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Calendar;
import java.util.Random;

public class SensorSource extends RichParallelSourceFunction<SensorReading> {

    private boolean running = true;

    @Override
    public void run(SourceContext<SensorReading> srcCtx) throws Exception {

        Random rand = new Random();

        String[] sensorIds = new String[10];
        double[] curFTemp = new double[10];
        for (int i = 0; i < 10; i++) {
            sensorIds[i] = "sensor_" + i;
            curFTemp[i] = 65 + (rand.nextGaussian() * 20);
        }

        while (running) {
            long curTime = Calendar.getInstance().getTimeInMillis();
            for (int i = 0; i < 10; i++) {
                curFTemp[i] += rand.nextGaussian() * 0.5;
                srcCtx.collect(new SensorReading(sensorIds[i], curTime, curFTemp[i]));
            }

            Thread.sleep(100);
        }
    }

    @Override
    public void cancel() {
        this.running = false;
    }
}

DataStream<SensorReading> sensorData = env.addSource(new SensorSource());

source.map(new MapFunction<Integer, Integer>() {
    @Override
    public Integer map(Integer value) throws Exception {
        return value + 1;
    }
});

dataStream.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public void flatMap(String value, Collector<String> out) throws Exception {
        for(String word: value.split(" ")){
            out.collect(word);
        }
    }
});

source.filter(new FilterFunction<String>() {  
	@Override  
	public boolean filter(String s) throws Exception {  
		return StringUtils.isNotBlank(s);  
	}
});

source.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {  
    @Override  
    public String getKey(Tuple2<String, Integer> value) throws Exception {  
        return value.f0;  
    }  
})

keyedStream.reduce(new ReduceFunction<Integer>() {
    @Override
    public Integer reduce(Integer value1, Integer value2) throws Exception {
        return value1 + value2;
    }
});

dataStream.union(stream1, stream2...);

connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() {
    @Override
    public Boolean map1(Integer value) {
        return true;
    }

    @Override
    public Boolean map2(String value) {
        return false;
    }
});

connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {
   @Override
   public void flatMap1(Integer value, Collector<String> out) {
       out.collect(value.toString());
   }

   @Override
   public void flatMap2(String value, Collector<String> out) {
       for (String word: value.split(",")) {
         out.collect(word);
       }
   }
});

source.shuffle();

source.rebalance();

source.rescale();

source.broadcast();

source.global();

dataStream.print();

<dependency>  
    <groupId>org.apache.flink</groupId>  
    <artifactId>flink-connector-kafka_2.12</artifactId>  
    <version>${flink.version}</version>  
</dependency>  
  
<dependency>  
    <groupId>org.apache.flink</groupId>  
    <artifactId>flink-connector-base</artifactId>  
    <version>${flink.version}</version>  
</dependency>

// 定义 kafka sink
Properties produceProperties = new Properties();
produceProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER_SERVERS);
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
    "test",
    new SimpleStringSchema(),
    produceProperties);

dataStream.addSink(kafkaProducer);

// Keyed Window
stream
       .keyBy(...)               <-  仅 keyed 窗口需要
       .window(...)              <-  必填项："assigner"
      [.trigger(...)]            <-  可选项："trigger" (省略则使用默认 trigger)
      [.evictor(...)]            <-  可选项："evictor" (省略则不使用 evictor)
      [.allowedLateness(...)]    <-  可选项："lateness" (省略则为 0)
      [.sideOutputLateData(...)] <-  可选项："output tag" (省略则不对迟到数据使用 side output)
       .reduce/aggregate/apply()      <-  必填项："function"
      [.getSideOutput(...)]      <-  可选项："output tag"

// Non-Keyed Window
stream
       .windowAll(...)           <-  不分组，将数据流中的所有元素分配到相应的窗口中
      [.trigger(...)]            <-  指定触发器Trigger（可选）
      [.evictor(...)]            <-  指定清除器Evictor(可选)
       .reduce/aggregate/process()      <-  窗口处理函数Window Function

// tumbling event-time windows  
input  
.keyBy(<key selector>)  
.window(TumblingEventTimeWindows.of(Time.seconds(5)))  
.<windowed transformation>(<window function>);

// sliding event-time windows  
input  
.keyBy(<key selector>)  
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))  
.<windowed transformation>(<window function>);

@PublicEvolving  
public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {  
	//创建初始累加器
    ACC createAccumulator();  

	//累加输入元素到累加器
    ACC add(IN value, ACC accumulator);  

	//从累加器提取输出结果
    OUT getResult(ACC accumulator);  

	//合并累加器
    ACC merge(ACC a, ACC b);  
}

public class AverageAccumulator {  
    long count;  
    long sum;  
}

public class WeightedAverage implements AggregateFunction<Double, AverageAccumulator, Double> {  
  
    public AverageAccumulator createAccumulator() {  
        return new AverageAccumulator();  
    }  
  
    public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {  
        a.count += b.count;  
        a.sum += b.sum;  
        return a;  
    }  
  
    public AverageAccumulator add(Double value, AverageAccumulator acc) {  
        acc.count += value;  
        acc.sum += value;  
        return acc;  
    }  
  
    public Double getResult(AverageAccumulator acc) {  
        return acc.sum / (double) acc.count;  
    }  
  
}

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<MyEvent> stream = env.readFile(
        myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
        FilePathFilter.createDefaultFilter(), typeInfo);

DataStream<MyEvent> withTimestampsAndWatermarks = stream
        .filter( event -> event.severity() == WARNING )
        .assignTimestampsAndWatermarks(<watermark strategy>);

withTimestampsAndWatermarks
        .keyBy( (event) -> event.getGroup() )
        .window(TumblingEventTimeWindows.of(Time.seconds(10)))
        .reduce( (a, b) -> a.add(b) )
        .addSink(...);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
env.enableCheckpointing(5000);

CheckpointConfig config = env.getCheckpointConfig();  
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);