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

This article mainly explains "how to use Flink Join". The content in the article is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn how to use Flink Join.

Join operator: two data streams pass through the same internal key partition, calculate the same key data elements of the two data streams in the window, and merge the output (similar to the inner join operation of the mysql table)

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

Join.java

Package com.flink.examples.functions;import com.flink.examples.DataSource;import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;import org.apache.flink.api.common.eventtime.WatermarkStrategy;import org.apache.flink.api.common.functions.FlatJoinFunction;import org.apache.flink.api.java.functions.KeySelector;import org.apache.flink.api.java.tuple.Tuple3;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStream Import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.util.Collector;import java.time.Duration;import java.util.Arrays;import java.util.List / * * @ Description Join operator: two data streams use the same internal key partition to calculate the same key data elements of two data streams in the window, and then merge the output (similar to the inner join operation of the mysql table) * / public class Join {/ * Flink supports two kinds of Join:Window Join (window join) and Interval Join (time interval join) This example demonstrates Window Join * official document: https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/stream/operators/joining.html * / * two data flow collections, inline the same key and assign them to the same window. Merge and print * @ param args * @ throws Exception * / public static void main (String [] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () Env.setParallelism (4); env.setStreamTimeCharacteristic (TimeCharacteristic.EventTime); / watermark automatic watermarking scheduling time / / env.getConfig () .setAutoWatermarkInterval (200); List tuple3List1 = DataSource.getTuple3ToList () List tuple3List2 = Arrays.asList (new Tuple3 ("Wu Qi", "girl", 18), new Tuple3 ("Wu Ba", "man", 30) / / Datastream 1 DataStream dataStream1 = env.fromCollection (tuple3List1) / / add watermark window. If not, the time window will wait for the watermark event time, and will not execute apply .assignTimestampsAndWatermarks (WatermarkStrategy.forBoundedOutOfOrderness (Duration.ofSeconds (2)). WithTimestampAssigner ((element, timestamp)-> System.currentTimeMillis () / / Datastream 2 DataStream dataStream2 = env.fromCollection (tuple3List2) / / add watermark window. If not, the time window will wait for the watermark event time. Apply .signTimestampsAndWatermarks (WatermarkStrategy.forBoundedOutOfOrderness (Duration.ofSeconds (2)) .withTimestampAssigner (new SerializableTimestampAssigner () {@ Override public long extractTimestamp (Tuple3 element, long timestamp) {return System.currentTimeMillis ()) will not be executed })) / / Datastream 3 DataStream newDataStream = dataStream1.join (dataStream2) .where (new KeySelector () {@ Override public String getKey (Tuple3 value) throws Exception {System.out.println ("first name:" + value.f0 + ", sex:" + value.f1); return value.f1 }}) .equalto (new KeySelector () {@ Override public String getKey (Tuple3 value) throws Exception {System.out.println ("second name:" + value.f0 + ", sex:" + value.f1); return value.f1 ) .window (TumblingEventTimeWindows.of (Time.seconds (1)) .apply (new FlatJoinFunction () {@ Override public void join (Tuple3 first, Tuple3 second) Collector out) throws Exception {out.collect (first.f0 + "|" + first.f1 + "|" + first.f2 + "|" + second.f0 + "|" + second.f1 + "|" + second.f2) ); newDataStream.print (); env.execute ("flink Join job");}}

Print the result

4 > Li Si | girl | 24 | Wu Qi | girl | 184 > Liu Liu | girl | 32 | Wu Qi | girl | 184 > Wu Qi | girl | 18 | Wu Qi | girl | 182 > Zhang San | man | 20 | Wu Ba | man | 302 > Wang Wu | man | 29 | Wu Ba | man | 302 > Wu Ba | man | 30 | man | 30 Thank you for your reading. The above is the content of "how to use Flink Join". After the study of this article, I believe you have a deeper understanding of how to use Flink Join, and the specific use needs to be verified by practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

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.