Runtime Context ctx

// 1. map 把String转换成长度生成 DataStream<Integer> mapStream = inputStream.map(new MapFunction<String, Integer>() { @Override public Integer map(String value) throws Exception { return value.length(); } }); // 2. flatmap 按逗号切分字段 DataStream<String> flatMapStream = inputStream.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { String[] fields=value.split(","); for (String field : fields){ out.collect(field); } } }); // 3.

前言 使用Flink处理数据时，可以基于Flink提供的批式处理（Batch Processing）和流式处理（Streaming Processing）API来实现，分别能够满足不同场景下应用数据的处理。这两种模式下，输入处理都被抽象为Source Operator，包含对应输入数据的处理逻辑；输出处理都被抽象为Sink Operator，包含了对应输出数据的处理逻辑。这里，我们只关注输出的Sink Operator实现。 Flink批式处理模式，运行Flink Batch Job时作用在有界的输入数据集上，所以Job运行的时间是有时限的，一旦Job运行完成，对应的整个数据处理应用就已经结束，比如，输入是一个数据文件，或者一个Hive SQL查询对应的结果集，等等。在批式处理模式下处理数据的输出时，主要需要实现一个自定义的OutputFormat，然后基于该OutputFormat来构建一个Sink，下面看下OutputFormat接口的定义，如下所示：...

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档 前言 Flink sink 流数据写入到es5和es7的简单示例。 一、写入到Elasticsearch5 pom maven依赖 <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-elasticsearch5_2.11</artifactId> <version>${flink.version}</version> </dependency> 代码如下（示例）： public class Es5SinkDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Row row=Row.of("张三","001",getTimestamp("2016-10-24...