实时计算Flink如何将字符转换为大写字母函数

Flink将字符转换为大写字母，需使用 UPPER函数，详情请参考官方文档。

方式一：直接调用原生的map算子

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource socketTextStream = env.socketTextStream("localhost", 8888);

//方式一：调用原生自带的map算子 SingleOutputStreamOperator wordSource = socketTextStream.map(word -> word.toUpperCase());

wordSource.print(); env.execute(); 方式二：调用底层的transform算子重定义实现

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource socketTextStream = env.socketTextStream("localhost", 8888); //方式二： SingleOutputStreamOperator wordSource = socketTextStream.transform("MyMap", TypeInformation.of(String.class), new StreamMap<>(String::toUpperCase));

wordSource.print(); env.execute(); 方式三：继承实现类自定义实现

/** * 类似于StreamMap操作 */ static class MyStreamMap extends AbstractStreamOperator implements OneInputStreamOperator<String, String> { @Override public void processElement(StreamRecord element) throws Exception { String elementValue = element.getValue(); output.collect(element.replace(elementValue.toUpperCase())); } }

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource socketTextStream = env.socketTextStream("localhost", 8888);

SingleOutputStreamOperator wordSource = socketTextStream.transform("MyStreamMap", TypeInformation.of(String.class), new MyStreamMap());

wordSource.print(); env.execute(); 方式四：实现RichMapFunction类

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource socketTextStream = env.socketTextStream("localhost", 8888);

SingleOutputStreamOperator wordSource = socketTextStream.map(new RichMapFunction<String, String>() { /** * 在构造对象之后，执行map方法之前执行一次 * 通常用于初始化工作，例如连接创建等 * @param parameters * @throws Exception */ @Override public void open(Configuration parameters) throws Exception { super.open(parameters); }

        /**
         * 在关闭subtask之前执行一次，例如做一些释放资源的工作
         * @throws Exception
         */
        @Override
        public void close() throws Exception {
            super.close();
        }

        @Override
        public String map(String s) throws Exception {
            int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();

            return indexOfThisSubtask + ":" + s.toUpperCase();
        }
    });

wordSource.print(); env.execute();