废弃

详见：

说明

用于处理数据，通过 DataStream 可以使用 map、flatMap、keyBy、filter、reduce 操作

例子

package top.malaoshi;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);//批模式
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//流模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断

        DataStream<String> dataStream = env.socketTextStream("hadoop1", 9999);

        SingleOutputStreamOperator<String> soso=dataStream.flatMap(new FlatMapImpl());
        SingleOutputStreamOperator<String> filterResult=soso.filter(new FilterImpl());
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapResult=filterResult.map(new MapImpl());
        KeyedStream<Tuple2<String, Integer>,String> ks=mapResult.keyBy(new keyByImpl());
//        SingleOutputStreamOperator<Tuple2<String, Integer>> soso3=ks.sum(1);//按照第二个元素聚合
        SingleOutputStreamOperator<Tuple2<String,Integer>> reduceResult=ks.reduce(new ReduceImpl());

        reduceResult.print();
        //启动并等待程序结束，可传入任务名
        env.execute("Window WordCount");
    }

    /**
     * 根据空格拆分字符串
     */
    public static class FlatMapImpl implements FlatMapFunction<String, String> {
        /**
         *
         * @param sentence 表示每一行数据
         * @param out  传出的集合
         */
        @Override
        public void flatMap(String sentence, Collector<String> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(word);
            }
        }
    }

    /*
    过滤掉lilei
     */
    public static class FilterImpl implements FilterFunction<String>{

        @Override
        public boolean filter(String s) throws Exception {
            if(s.equals("lilei")) {
                return false;
            }else{
                return true;
            }
        }
    }

    /*
    将集合中的元素（单词），改成：(单词,1)
     */
    public static class MapImpl implements MapFunction<String,Tuple2<String, Integer>>{
        @Override
        public Tuple2<String, Integer> map(String s) throws Exception {
            return Tuple2.of(s,1);
        }
    }

    /**
     集合中的元素是 Tuple：(单词,1)，根据第一个元素（即：单词）进行分组
     */
    public static class keyByImpl implements KeySelector<Tuple2<String, Integer>,String>{
        /**
         *
         * @param t：传入的元祖，格式：(单词,1)
         * @return
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Integer> t) throws Exception {
            return t.f0;
        }
    }

    /*
    聚合
     */
    public static class ReduceImpl implements ReduceFunction<Tuple2<String, Integer>>{
        /**
         *
         * @param t1 上一次计算的结果
         * @param t2 本次的元素
         * @return
         * @throws Exception
         */
        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
            return Tuple2.of(t1.f0,t1.f1+t2.f1);
        }
    }
}

测试

需要启动 socket TCP 服务器

nc -l 9999

需要先安装 nc，详见链接

注意：要先启动 socket服务端，因为启动 java 要连接服务器，连不上会报错

执行java代码

输入数据：

在 nc 环境下，分别输入：

lilei

hanmeimei

hanmeimei lucy

lucy lilei

在idea控制台中，显示：

6> (hanmeimei,1)
4> (lucy,1)
6> (hanmeimei,2)
4> (lucy,2)

lilei被过滤掉了