<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
  <version>${flink.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime-web</artifactId>
  <version>${flink.version}</version>
</dependency>

Configuration conf = new Configuration();
//Устанавливаем локальный порт, привязанный к WebUI
conf.setString(RestOptions.BIND_PORT,"8081");
//Используем конфигурацию
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

val configuration = new Configuration()
//Устанавливаем локальный порт, привязанный к WebUI
configuration.set(RestOptions.BIND_PORT,"8081")
//Используем конфигурацию
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration)

//1. Использовать локальный режим.
Configuration conf = new Configuration();
//Устанавливаем локальный порт, привязанный к WebUI
conf.setString(RestOptions.BIND_PORT,"8081");
//Используем конфигурацию
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

//2. Чтение данных сокета
DataStreamSource<String> ds = env.socketTextStream("node3", 9999);

//3. Подготовьте данные формата K, V.
SingleOutputStreamOperator<Tuple2<String, Integer>> tupleDS = ds.flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
    String[] words = line.split(",");
    for (String word : words) {
        out.collect(Tuple2.of(word, 1));
    }
}).returns(Types.TUPLE(Types.STRING, Types.INT));

//4. Совокупные результаты печати.
tupleDS.keyBy(tp -> tp.f0).sum(1).print();

//5.execute запускает выполнение
env.execute();

//1. Создаём локальную среду WebUI.
val configuration = new Configuration()
//Устанавливаем привязанный локальный порт
configuration.set(RestOptions.BIND_PORT,"80")
//Первый метод настройки
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration)

//2.Scala Неявное преобразование импорта потоковой обработки с использованием Scala API Неявное преобразование необходимо для вывода типа после операции функции.
import org.apache.flink.streaming.api.scala._

//3. Чтение данных сокета
val linesDS: DataStream[String] = env.socketTextStream("node3", 9999)

//4. Выполнение статистики WordCount.
linesDS.flatMap(line=>{line.split(",")})
  .map((_,1))
  .keyBy(_._1)
  .sum(1)
  .print()

//5.Наконец используем выполнение Метод запускает выполнение
env.execute()