Shulou(Shulou.com)06/01 Report--

This article mainly introduces "how to use Flink SideOutput". In daily operation, I believe many people have doubts about how to use Flink SideOutput. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts about "how to use Flink SideOutput". Next, please follow the editor to study!

SideOutput method: (side output) get additional output streams from the main data stream according to outputTag (used in shunt scenarios)

Sample environment

Java.version: 1.8.xflink.version: 1.11.1

Sample data source (project code cloud download)

Building Development Environment and data of Flink system example

SideOutput.java

Import com.flink.examples.DataSource;import org.apache.flink.api.java.tuple.Tuple3;import org.apache.flink.api.java.tuple.Tuple4;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.ProcessFunction;import org.apache.flink.util.Collector;import org.apache.flink.util.OutputTag;import java.util.List / * * @ Description SideOutput method: (side output) get additional output streams from the main data stream according to outputTag (for use in shunt scenarios) * / public class SideOutput {/ * traverse the collection, split the data stream into multiple streams and print * @ param args * @ throws Exception * / public static void main (String [] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () Env.setParallelism (1); List tuple3List = DataSource.getTuple3ToList (); / / Datastream DataStream dataStream = env.fromCollection (tuple3List); / / Custom man and girl gender tag OutputTag manTag = new OutputTag ("man") {}; OutputTag girlTag = new OutputTag ("girl") {} / / input all stream data to process for processing SingleOutputStreamOperator output = dataStream.process (new ProcessFunction () {@ Override public void processElement (Tuple3 value, Context ctx, Collector out) throws Exception {/ / divide the data stream by gender and create a new Tuple4 Bind to tag if of different genders (value.f1.equals ("man")) {ctx.output (manTag, Tuple4.of (value.f0, value.f1, value.f2, "male")) } else {ctx.output (girlTag, Tuple4.of (value.f0, value.f1, value.f2, "female");}); / / get the data flow of the specified tag DataStream dataStream1 = output.getSideOutput (manTag); DataStream dataStream2 = output.getSideOutput (girlTag) / / print dataStream1.print (); dataStream2.print (); env.execute ("flink Split job");}}

Print the result

(Zhang San, man,20, male) (Li Si, girl,24, female) (Wang Wu, man,29, male) (Liu Liu, girl,32, female) (Wu Qi, girl,18, female) (Wu Ba, man,30, male) at this point, the study on "how to use Flink SideOutput" is over. I hope to solve everyone's doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.