env.enableCheckpointing(1000)

/user-defined-checkpoint-dir

    /{job-id}

        |

        + --shared/

        + --taskowned/

        + --chk-1/

        + --chk-2/

        + --chk-3/

        ...

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> text = env.readTextFile("file:///path/to/file");

DataStream<String> input = ...;

DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {

    @Override

    public Integer map(String value) {

        return Integer.parseInt(value);

    }

});

writeAsText(String path)

print()

final JobClient jobClient = env.executeAsync();

final JobExecutionResult jobExecutionResult = jobClient.getJobExecutionResult().get();

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.util.Collector;

public class WindowWordCount {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> dataStream = env

                .socketTextStream("localhost", 9999)

                .flatMap(new Splitter())

                .keyBy(value -> value.f0)

                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

                .sum(1);

        dataStream.print();

        env.execute("Window WordCount");

    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override

        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {

            for (String word: sentence.split(" ")) {

                out.collect(new Tuple2<String, Integer>(word, 1));

            }

        }

    }

}

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// read text file from local files system

DataSet<String> localLines = env.readTextFile("file:///path/to/my/textfile");

// read text file from an HDFS running at nnHost:nnPort

DataSet<String> hdfsLines = env.readTextFile("hdfs://nnHost:nnPort/path/to/my/textfile");

// read a CSV file with three fields

DataSet<Tuple3<Integer, String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")

                           .types(Integer.class, String.class, Double.class);

// read a CSV file with five fields, taking only two of them

DataSet<Tuple2<String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")

                               .includeFields("10010")  // take the first and the fourth field

                           .types(String.class, Double.class);

// read a CSV file with three fields into a POJO (Person.class) with corresponding fields

DataSet<Person>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")

                         .pojoType(Person.class, "name", "age", "zipcode");

// read a file from the specified path of type SequenceFileInputFormat

DataSet<Tuple2<IntWritable, Text>> tuples =

 env.createInput(HadoopInputs.readSequenceFile(IntWritable.class, Text.class, "hdfs://nnHost:nnPort/path/to/file"));

// creates a set from some given elements

DataSet<String> value = env.fromElements("Foo", "bar", "foobar", "fubar");

// generate a number sequence

DataSet<Long> numbers = env.generateSequence(1, 10000000);

// Read data from a relational database using the JDBC input format

DataSet<Tuple2<String, Integer> dbData =

    env.createInput(

      JdbcInputFormat.buildJdbcInputFormat()

                     .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")

                     .setDBUrl("jdbc:derby:memory:persons")

                     .setQuery("select name, age from persons")

                     .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO))

                     .finish()

    );

// Note: Flink's program compiler needs to infer the data types of the data items which are returned

// by an InputFormat. If this information cannot be automatically inferred, it is necessary to

// manually provide the type information as shown in the examples above.

// enable recursive enumeration of nested input files

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// create a configuration object

Configuration parameters = new Configuration();

// set the recursive enumeration parameter

parameters.setBoolean("recursive.file.enumeration", true);

// pass the configuration to the data source

DataSet<String> logs = env.readTextFile("file:///path/with.nested/files")

              .withParameters(parameters);

// text data

DataSet<String> textData = // [...]

// write DataSet to a file on the local file system

textData.writeAsText("file:///my/result/on/localFS");

// write DataSet to a file on an HDFS with a namenode running at nnHost:nnPort

textData.writeAsText("hdfs://nnHost:nnPort/my/result/on/localFS");

// write DataSet to a file and overwrite the file if it exists

textData.writeAsText("file:///my/result/on/localFS", WriteMode.OVERWRITE);

// tuples as lines with pipe as the separator "a|b|c"

DataSet<Tuple3<String, Integer, Double>> values = // [...]

values.writeAsCsv("file:///path/to/the/result/file", "\n", "|");

// this writes tuples in the text formatting "(a, b, c)", rather than as CSV lines

values.writeAsText("file:///path/to/the/result/file");

// this writes values as strings using a user-defined TextFormatter object

values.writeAsFormattedText("file:///path/to/the/result/file",

    new TextFormatter<Tuple2<Integer, Integer>>() {

        public String format (Tuple2<Integer, Integer> value) {

            return value.f1 + " - " + value.f0;

        }

    });

DataSet<Tuple3<String, Integer, Double>> myResult = [...]

// write Tuple DataSet to a relational database

myResult.output(

    // build and configure OutputFormat

    JdbcOutputFormat.buildJdbcOutputFormat()

                    .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")

                    .setDBUrl("jdbc:derby:memory:persons")

                    .setQuery("insert into persons (name, age, height) values (?,?,?)")

                    .finish()

    );

DataSet<Tuple3<Integer, String, Double>> tData = // [...]

DataSet<Tuple2<BookPojo, Double>> pData = // [...]

DataSet<String> sData = // [...]

// sort output on String field in ascending order

tData.sortPartition(1, Order.ASCENDING).print();

// sort output on Double field in descending and Integer field in ascending order

tData.sortPartition(2, Order.DESCENDING).sortPartition(0, Order.ASCENDING).print();

// sort output on the "author" field of nested BookPojo in descending order

pData.sortPartition("f0.author", Order.DESCENDING).writeAsText(...);

// sort output on the full tuple in ascending order

tData.sortPartition("*", Order.ASCENDING).writeAsCsv(...);

// sort atomic type (String) output in descending order

sData.sortPartition("*", Order.DESCENDING).writeAsText(...);

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

env.setBufferTimeout(timeoutMillis);//預設是100ms

env.generateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

// some ordinary POJO

public class WC {

  public String word;

  public int count;

  public String getWord() { return word; }

}

DataStream<WC> words = // [...]

KeyedStream<WC> keyed = words

  .keyBy(WC::getWord);

public class WordCountExample {

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

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<String> text = env.fromElements(

            "Who's there?",

            "I think I hear them. Stand, ho! Who's there?");

        DataSet<Tuple2<String, Integer>> wordCounts = text

            .flatMap(new LineSplitter())

            .groupBy(0)

            .sum(1);

        wordCounts.print();

    }

    public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override

        public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {

            for (String word : line.split(" ")) {

                out.collect(new Tuple2<String, Integer>(word, 1));

            }

        }

    }

}

DataSet<...> input = // [...]

DataSet<...> reduced = input

  .groupBy(/*define key here*/)

  .reduceGroup(/*do something*/);

class MyMapFunction implements MapFunction<String, Integer> {

  public Integer map(String value) { return Integer.parseInt(value); }

};

data.map(new MyMapFunction());

data.map(new MapFunction<String, Integer> () {

  public Integer map(String value) { return Integer.parseInt(value); }

});

data.filter(s -> s.startsWith("http://"));

data.reduce((i1,i2) -> i1 + i2);

class MyMapFunction implements MapFunction<String, Integer> {

  public Integer map(String value) { return Integer.parseInt(value); }

};

class MyMapFunction extends RichMapFunction<String, Integer> {

  public Integer map(String value) { return Integer.parseInt(value); }

};

data.map(new MyMapFunction());

data.map (new RichMapFunction<String, Integer>() {

  public Integer map(String value) { return Integer.parseInt(value); }

});

private IntCounter numLines = new IntCounter();

getRuntimeContext().addAccumulator("num-lines", this.numLines);

this.numLines.add(1);

myJobExecutionResult.getAccumulatorResult("num-lines")

DataStream<T> input = ...;

// tumbling event-time windows

input

    .keyBy(<key selector>)

    .window(TumblingEventTimeWindows.of(Time.seconds(5)))

    .<windowed transformation>(<window function>);

// tumbling processing-time windows

input

    .keyBy(<key selector>)

    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

    .<windowed transformation>(<window function>);

// daily tumbling event-time windows offset by -8 hours.

input

    .keyBy(<key selector>)

    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))

    .<windowed transformation>(<window function>);

DataStream<T> input = ...;

// sliding event-time windows

input

    .keyBy(<key selector>)

    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))

    .<windowed transformation>(<window function>);

// sliding processing-time windows

input

    .keyBy(<key selector>)

    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))

    .<windowed transformation>(<window function>);

// sliding processing-time windows offset by -8 hours

input

    .keyBy(<key selector>)

    .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))

    .<windowed transformation>(<window function>);

DataStream<T> input = ...;

// event-time session windows with static gap

input

    .keyBy(<key selector>)

    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))

    .<windowed transformation>(<window function>);

// event-time session windows with dynamic gap

input

    .keyBy(<key selector>)

    .window(EventTimeSessionWindows.withDynamicGap((element) -> {

        // determine and return session gap

    }))

    .<windowed transformation>(<window function>);

// processing-time session windows with static gap

input

    .keyBy(<key selector>)

    .window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))

    .<windowed transformation>(<window function>);

// processing-time session windows with dynamic gap

input

    .keyBy(<key selector>)

    .window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {

        // determine and return session gap

    }))

    .<windowed transformation>(<window function>);

DataStream<T> input = ...;

input

    .keyBy(<key selector>)

    .window(GlobalWindows.create())

    .<windowed transformation>(<window function>);

DataStream<Tuple2<String, Long>> input = ...;

input

    .keyBy(<key selector>)

    .window(<window assigner>)

    .reduce(new ReduceFunction<Tuple2<String, Long>> {

      public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) {

        return new Tuple2<>(v1.f0, v1.f1 + v2.f1);

      }

    });

/**

 * The accumulator is used to keep a running sum and a count. The {@code getResult} method

 * computes the average.

 */

private static class AverageAggregate

    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {

  @Override

  public Tuple2<Long, Long> createAccumulator() {

    return new Tuple2<>(0L, 0L);

  }

  @Override

  public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {

    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);

  }

  @Override

  public Double getResult(Tuple2<Long, Long> accumulator) {

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

  }

  @Override

  public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {

    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);

  }

}

DataStream<Tuple2<String, Long>> input = ...;

input

    .keyBy(<key selector>)

    .window(<window assigner>)

    .aggregate(new AverageAggregate());

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> implements Function {

    /**

     * Evaluates the window and outputs none or several elements.

     *

     * @param key The key for which this window is evaluated.

     * @param context The context in which the window is being evaluated.

     * @param elements The elements in the window being evaluated.

     * @param out A collector for emitting elements.

     *

     * @throws Exception The function may throw exceptions to fail the program and trigger recovery.

     */

    public abstract void process(

            KEY key,

            Context context,

            Iterable<IN> elements,

            Collector<OUT> out) throws Exception;

       /**

        * The context holding window metadata.

        */

       public abstract class Context implements java.io.Serializable {

           /**

            * Returns the window that is being evaluated.

            */

           public abstract W window();

           /** Returns the current processing time. */

           public abstract long currentProcessingTime();

           /** Returns the current event-time watermark. */

           public abstract long currentWatermark();

           /**

            * State accessor for per-key and per-window state.

            *

            * <p><b>NOTE:</b>If you use per-window state you have to ensure that you clean it up

            * by implementing {@link ProcessWindowFunction#clear(Context)}.

            */

           public abstract KeyedStateStore windowState();

           /**

            * State accessor for per-key global state.

            */

           public abstract KeyedStateStore globalState();

       }

}

DataStream<Tuple2<String, Long>> input = ...;

input

  .keyBy(t -> t.f0)

  .window(TumblingEventTimeWindows.of(Time.minutes(5)))

  .process(new MyProcessWindowFunction());

/* ... */

public class MyProcessWindowFunction

    extends ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow> {

  @Override

  public void process(String key, Context context, Iterable<Tuple2<String, Long>> input, Collector<String> out) {

    long count = 0;

    for (Tuple2<String, Long> in: input) {

      count++;

    }

    out.collect("Window: " + context.window() + "count: " + count);

  }

}

DataStream<SensorReading> input = ...;

input

  .keyBy(<key selector>)

  .window(<window assigner>)

  .reduce(new MyReduceFunction(), new MyProcessWindowFunction());

// Function definitions

private static class MyReduceFunction implements ReduceFunction<SensorReading> {

  public SensorReading reduce(SensorReading r1, SensorReading r2) {

      return r1.value() > r2.value() ? r2 : r1;

  }

}

private static class MyProcessWindowFunction

    extends ProcessWindowFunction<SensorReading, Tuple2<Long, SensorReading>, String, TimeWindow> {

  public void process(String key,

                    Context context,

                    Iterable<SensorReading> minReadings,

                    Collector<Tuple2<Long, SensorReading>> out) {

      SensorReading min = minReadings.iterator().next();

      out.collect(new Tuple2<Long, SensorReading>(context.window().getStart(), min));

  }

}

DataStream<Tuple2<String, Long>> input = ...;

input

  .keyBy(<key selector>)

  .window(<window assigner>)

  .aggregate(new AverageAggregate(), new MyProcessWindowFunction());

// Function definitions

/**

 * The accumulator is used to keep a running sum and a count. The {@code getResult} method

 * computes the average.

 */

private static class AverageAggregate

    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {

  @Override

  public Tuple2<Long, Long> createAccumulator() {

    return new Tuple2<>(0L, 0L);

  }

  @Override

  public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {

    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);

  }

  @Override

  public Double getResult(Tuple2<Long, Long> accumulator) {

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

  }

  @Override

  public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {

    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);

  }

}

private static class MyProcessWindowFunction

    extends ProcessWindowFunction<Double, Tuple2<String, Double>, String, TimeWindow> {

  public void process(String key,

                    Context context,

                    Iterable<Double> averages,

                    Collector<Tuple2<String, Double>> out) {

      Double average = averages.iterator().next();

      out.collect(new Tuple2<>(key, average));

  }

}

DataStream<T> input = ...;

input

    .keyBy(<key selector>)

    .window(<window assigner>)

    .allowedLateness(<time>)

    .<windowed transformation>(<window function>);

final OutputTag<T> lateOutputTag = new OutputTag<T>("late-data"){};

DataStream<T> input = ...;

SingleOutputStreamOperator<T> result = input

    .keyBy(<key selector>)

    .window(<window assigner>)

    .allowedLateness(<time>)

    .sideOutputLateData(lateOutputTag)

    .<windowed transformation>(<window function>);

DataStream<T> lateStream = result.getSideOutput(lateOutputTag);

// 這需要是一個匿名的內部類，以便我們分析型別

OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

DataStream<Integer> input = ...;

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = input

  .process(new ProcessFunction<Integer, Integer>() {

      @Override

      public void processElement(

          Integer value,

          Context ctx,

          Collector<Integer> out) throws Exception {

        // emit data to regular output

        out.collect(value);

        // emit data to side output

        ctx.output(outputTag, "sideout-" + String.valueOf(value));

      }

    });

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = ...;

DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);