```java package com.gosuncn; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.typeinfo.Types; 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.util.Collector; public class WordCountStreamingJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<String> lines = env.socketTextStream("47.52.74.183", 8888); DataStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String line, Collector<String> out) throws Exception { String[] words = line.split(" "); for (String word : words) { out.collect(word); } } }).returns(Types.STRING); DataStream<Tuple2> wordAndOne = words.map(new MapFunction<String, Tuple2>() { @Override public Tuple2 map(String word) throws Exception { return Tuple2.of(word, 1); } }).returns(Types.TUPLE(Types.STRING, Types.INT)); DataStream<Tuple2> summed = wordAndOne.keyBy(0).sum(1); summed.print(); env.execute("WordCountStreamingJob"); } } ```