// Установка атрибута Ingestion Time для StreamExecutionEnvironment objectstreamEnv.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

// будет использоваться с потоком, не снабженным ключами
public AllWindowedStream<T,TimeWindow> timeWindowAll(Time size)
// Кувыркающееся окно для потока, снабженного ключами
public WindowedStream<T,KEY,TimeWindow> timeWindow(Time size)

// скользящее окно длиной 1 минуту и с интервалом срабатывания 30 секунд
dataStreamObject.timeWindow(Time.minutes(1), Time.seconds(30))

// Определение фиксированного сеансового окна длительностью 2 секунды
dataStreamObject.window(ProcessingTimeSessionWindows.withGap(Time.seconds(2)))
// Определение динамического сеансового окна, которое может быть задано элементами потока
dataStreamObject.window(EventTimeSessionWindows.withDynamicGap((elem) -> {
        // возвращается промежуток между сеансами, который может зависеть от событий потока
    }))

// Каждая запись преобразуется в объект InputData; каждая новая строка считается новой записью
DataStream<InputData> inputDataObjectStream
          = dataStream
           .map((MapFunction<String, InputData>) inputStr -> {
                System.out.println("--- Received Record : " + inputStr);
                return InputData.getDataObject(inputStr);
          });

// Каждая запись преобразуется в кортеж с именем и счетом
        DataStream<Tuple2<String, Integer>> userCounts
                = inputDataObjectStream
                .map(new MapFunction<InputData,Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String,Integer> map(InputData item) {
                        return new Tuple2<String,Integer>(item.getName() ,item.getScore() );
                    }
                })
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)  // возвращает KeyedStream<T, Tuple> на основе первого элемента (поля 'name')
                //.timeWindowAll(Time.seconds(windowInterval)) // НЕ ИСПОЛЬЗОВАТЬ timeWindowAll с потоком на основе ключей
                .timeWindow(Time.seconds(2)) // вернуть WindowedStream<T, KEY, TimeWindow>
                .reduce((x,y) -> new Tuple2<String,Integer>( x.f0+"-"+y.f0, x.f1+y.f1));

// Определить временное окно и подсчитать количество записей
           DataStream<Tuple2<String,Integer>> inputCountSummary
                    = inputDataObjectStream
                    .map( item
                            -> new Tuple2<String,Integer>
                            (String.valueOf(System.currentTimeMillis()),1))
// для каждого элемента вернуть кортеж из временной метки и целого числа (1)
                    .returns(Types.TUPLE(Types.STRING ,Types.INT))
                    .timeWindowAll(Time.seconds(windowInterval)) // кувыркающееся окно
                    .reduce((x,y) -> // суммируем числа, и так до достижения единого результата
                            (new Tuple2<String, Integer>(x.f0, x.f1 + y.f1)));
            // Задаем в качестве стока для потокового файла каталог вывода
            final StreamingFileSink<Tuple2<String,Integer>> countSink
                    = StreamingFileSink
                        .forRowFormat(new Path(outputDir),
                                new SimpleStringEncoder<Tuple2<String,Integer>>
                                        ("UTF-8"))
                        .build();
            // Добавляем поток стока к DataStream; при таком условии inputCountSummary будет вписан в путь countSink
            inputCountSummary.addSink(countSink);

// Определить отдельный поток для Исполнителей
            final OutputTag<Tuple2<String,String>> playerTag
                    = new OutputTag<Tuple2<String,String>>("player"){};
            // Определить отдельный поток для Певцов
            final OutputTag<Tuple2<String,Integer>> singerTag
                    = new OutputTag<Tuple2<String,Integer>>("singer"){};
            // Преобразовать каждую запись в объект InputData и разделить главный поток на два боковых.
            SingleOutputStreamOperator<InputData> inputDataMain
                    = inputStream
                    .process(new ProcessFunction<String, InputData>() {
                        @Override
                        public void processElement(
                                String inputStr,
                                Context ctx,
                                Collector<InputData> collInputData) {
                            Utils.print(Utils.COLOR_CYAN, "Received record : " + inputStr);
                            // Преобразовать строку в объект InputData
                            InputData inputData = InputData.getDataObject(inputStr);
                            switch (inputData.getType())
                            {
                                case "Singer":
// Создать выходной кортеж со значениями имени и счета
                                    ctx.output(singerTag,
                                            new Tuple2<String,Integer>
                                                    (inputData.getName(), inputData.getScore()));
                                    break;
                                case "Player":
// Создать выходной кортеж со значениями имени и типа;
// Если новоиспеченный кортеж не совпадает с типом playerTag, то выбрасывается ошибка компиляции ("вывод метода не может быть применен к указанным типам")
                                    ctx.output(playerTag,
                                            new Tuple2<String, String>
                                                    (inputData.getName(), inputData.getType()));
                                    break;
                                default:
                      // Собрать вывод основного потока как объекты InputData
                                    collInputData.collect(inputData);
                                    break;
                            }
                        }
                    });

// В описании возвращенного потока учтены типы данных обоих потоков
        ConnectedStreams<Tuple2<String, Integer>, Tuple2<String, String>> mergedStream
                = singerStream
                .connect(playerStream);
        DataStream<Tuple4<String, String, String, Integer>> combinedStream
                = mergedStream.map(new CoMapFunction<
                        Tuple2<String, Integer>, // Поток 1
                        Tuple2<String, String>, // Поток 2
                        Tuple4<String, String, String, Integer> //Вывод
                        >() {
                            @Override
                            public Tuple4<String, String, String, Integer>  //Обработка потока 1
                            map1(Tuple2<String, Integer> singer) throws Exception {
                                return new Tuple4<String, String, String, Integer>
                                        ("Source: singer stream", singer.f0, "", singer.f1);
                            }
                            @Override
                            public Tuple4<String, String, String, Integer>
// Обработка потока 2
                            map2(Tuple2<String, String> player) throws Exception {
                                return new Tuple4<String, String, String, Integer>
                                        ("Source: player stream", player.f0, player.f1, 0);
                            }
                 });