常见的维表Join方式有四种：

  • 预加载维表
  • 热存储维表
  • 广播维表
  • Temporal table function join

下面分别使用这四种方式来实现一个join的需求，这个需求是：
一个主流中数据是用户信息，字段包括用户姓名、城市id；
维表是城市数据，字段包括城市ID、城市名称。
要求用户表与城市表关联，输出为：用户名称、城市ID、城市名称。
用户表表结构如下：
Flink维表Join实践 - 图1
城市维表表结构如下：
Flink维表Join实践 - 图2

1. 预加载维表

通过定义一个类实现RichMapFunction，在open()中读取维表数据加载到内存中，在probe流map()方法中与维表数据进行关联。RichMapFunction中open方法里加载维表数据到内存的方式特点如下：

优点：实现简单。 缺点：因为数据存于内存，所以只适合小数据量并且维表数据更新频率不高的情况下。虽然可以在open中定义一个定时器定时更新维表，但是还是存在维表更新不及时的情况。

示例：

  1. package join;
  3. import org.apache.flink.api.common.functions.RichMapFunction;
  4. import org.apache.flink.api.common.typeinfo.TypeHint;
  5. import org.apache.flink.api.java.tuple.Tuple2;
  6. import org.apache.flink.api.java.tuple.Tuple3;
  7. import org.apache.flink.configuration.Configuration;
  8. import org.apache.flink.streaming.api.datastream.DataStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import java.util.HashMap;
  11. import java.util.Map;
  13. /**
  14.  * 这个例子是从socket中读取的流，数据为用户名称和城市id，维表是城市id、城市名称，
  15.  * 主流和维表关联，得到用户名称、城市id、城市名称
  16.  * 这个例子采用在RichMapfunction类的open方法中将维表数据加载到内存
  17.  **/
  18. public class JoinDemo1 {
  19.     public static void main(String[] args) throws Exception {
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         DataStream<Tuple2<String, Integer>> textStream = env.socketTextStream("localhost", 9000, "\n")
  22.                 .map(p -> {
  23.                     //输入格式为：user,1000,分别是用户名称和城市编号
  24.                     String[] list = p.split(",");
  25.                     return new Tuple2<String, Integer>(list[0], Integer.valueOf(list[1]));
  26.                 })
  27.                 .returns(new TypeHint<Tuple2<String, Integer>>() {
  28.                 });
  29.         DataStream<Tuple3<String, Integer, String>> result = textStream.map(new MapJoinDemo1());
  30.         result.print();
  31.         env.execute("joinDemo1");
  32.     }
  34.     static class MapJoinDemo1 extends RichMapFunction<Tuple2<String, Integer>, Tuple3<String, Integer, String>> {
  35.         //定义一个变量，用于保存维表数据在内存
  36.         Map<Integer, String> dim;
  38.         @Override
  39.         public void open(Configuration parameters) throws Exception {
  40.             //在open方法中读取维表数据，可以从数据中读取、文件中读取、接口中读取等等。
  41.             dim = new HashMap<>();
  42.             dim.put(1001, "beijing");
  43.             dim.put(1002, "shanghai");
  44.             dim.put(1003, "wuhan");
  45.             dim.put(1004, "changsha");
  46.         }
  48.         @Override
  49.         public Tuple3<String, Integer, String> map(Tuple2<String, Integer> value) throws Exception {
  50.             //在map方法中进行主流和维表的关联
  51.             String cityName = "";
  52.             if (dim.containsKey(value.f1)) {
  53.                 cityName = dim.get(value.f1);
  54.             }
  55.             return new Tuple3<>(value.f0, value.f1, cityName);
  56.         }
  57.     }
  58. }

2. 热存储维表

这种方式是将维表数据存储在Redis、HBase、MySQL等外部存储中，实时流在关联维表数据的时候实时去外部存储中查询，这种方式特点如下：

优点：维度数据量不受内存限制，可以存储很大的数据量。 缺点：因为维表数据在外部存储中，读取速度受制于外部存储的读取速度；另外维表的同步也有延迟。

2.1 使用cache来减轻访问压力

可以使用缓存来存储一部分常访问的维表数据，以减少访问外部系统的次数，比如使用guava Cache。
下面是一个例子：

  1. package join;
  3. import com.google.common.cache.*;
  4. import org.apache.flink.api.common.functions.RichMapFunction;
  5. import org.apache.flink.api.common.typeinfo.TypeHint;
  6. import org.apache.flink.api.java.tuple.Tuple2;
  7. import org.apache.flink.api.java.tuple.Tuple3;
  8. import org.apache.flink.configuration.Configuration;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import java.util.HashMap;
  13. import java.util.Map;
  14. import java.util.concurrent.TimeUnit;
  16. public class JoinDemo2 {
  17.     public static void main(String[] args) throws Exception {
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         DataStream<Tuple2<String, Integer>> textStream = env.socketTextStream("localhost", 9000, "\n")
  21.                 .map(p -> {
  22.                     //输入格式为：user,1000,分别是用户名称和城市编号
  23.                     String[] list = p.split(",");
  24.                     return new Tuple2<String, Integer>(list[0], Integer.valueOf(list[1]));
  25.                 })
  26.                 .returns(new TypeHint<Tuple2<String, Integer>>() {
  27.                 });
  29.         DataStream<Tuple3<String, Integer, String>> result = textStream.map(new MapJoinDemo1());
  30.         result.print();
  31.         env.execute("joinDemo1");
  32.     }
  34.     static class MapJoinDemo1 extends RichMapFunction<Tuple2<String, Integer>, Tuple3<String, Integer, String>> {
  35.         LoadingCache<Integer, String> dim;
  37.         @Override
  38.         public void open(Configuration parameters) throws Exception {
  39.             //使用google LoadingCache来进行缓存
  40.             dim = CacheBuilder.newBuilder()
  41.                     //最多缓存个数，超过了就根据最近最少使用算法来移除缓存
  42.                     .maximumSize(1000)
  43.                     //在更新后的指定时间后就回收
  44.                     .expireAfterWrite(10, TimeUnit.MINUTES)
  45.                     //指定移除通知
  46.                     .removalListener(new RemovalListener<Integer, String>() {
  47.                         @Override
  48.                         public void onRemoval(RemovalNotification<Integer, String> removalNotification) {
  49.                             System.out.println(removalNotification.getKey() + "被移除了，值为：" + removalNotification.getValue());
  50.                         }
  51.                     })
  52.                     .build(
  53.                             //指定加载缓存的逻辑
  54.                             new CacheLoader<Integer, String>() {
  55.                                 @Override
  56.                                 public String load(Integer cityId) throws Exception {
  57.                                     String cityName = readFromHbase(cityId);
  58.                                     return cityName;
  59.                                 }
  60.                             }
  61.                     );
  63.         }
  65.         private String readFromHbase(Integer cityId) {
  66.             //读取hbase
  67.             //这里写死，模拟从hbase读取数据
  68.             Map<Integer, String> temp = new HashMap<>();
  69.             temp.put(1001, "beijing");
  70.             temp.put(1002, "shanghai");
  71.             temp.put(1003, "wuhan");
  72.             temp.put(1004, "changsha");
  73.             String cityName = "";
  74.             if (temp.containsKey(cityId)) {
  75.                 cityName = temp.get(cityId);
  76.             }
  78.             return cityName;
  79.         }
  81.         @Override
  82.         public Tuple3<String, Integer, String> map(Tuple2<String, Integer> value) throws Exception {
  83.             //在map方法中进行主流和维表的关联
  84.             String cityName = "";
  85.             if (dim.get(value.f1) != null) {
  86.                 cityName = dim.get(value.f1);
  87.             }
  88.             return new Tuple3<>(value.f0, value.f1, cityName);
  89.         }
  90.     }
  91. }

2.2 使用异步IO来提高访问吞吐量

Flink与外部存储系统进行读写操作的时候可以使用同步方式，也就是发送一个请求后等待外部系统响应，然后再发送第二个读写请求，这样的方式吞吐量比较低，可以用提高并行度的方式来提高吞吐量，但是并行度多了也就导致了进程数量多了，占用了大量的资源。
Flink中可以使用异步IO来读写外部系统，这要求外部系统客户端支持异步IO，不过目前很多系统都支持异步IO客户端。但是如果使用异步就要涉及到三个问题：

  • 超时：如果查询超时那么就认为是读写失败，需要按失败处理；
  • 并发数量：如果并发数量太多，就要触发Flink的反压机制来抑制上游的写入;
  • 返回顺序错乱：顺序错乱了要根据实际情况来处理，Flink支持两种方式：允许乱序、保证顺序。

Flink维表Join实践 - 图3
下面是一个实例，演示了试用异步IO来访问维表：

  1. package join;
  3. import org.apache.flink.api.common.typeinfo.TypeHint;
  4. import org.apache.flink.api.java.tuple.Tuple2;
  5. import org.apache.flink.api.java.tuple.Tuple3;
  6. import org.apache.flink.configuration.Configuration;
  7. import org.apache.flink.streaming.api.datastream.AsyncDataStream;
  8. import org.apache.flink.streaming.api.datastream.DataStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.streaming.api.functions.async.ResultFuture;
  11. import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
  12. import java.sql.DriverManager;
  13. import java.sql.PreparedStatement;
  14. import java.sql.ResultSet;
  15. import java.util.ArrayList;
  16. import java.util.List;
  17. import java.util.concurrent.TimeUnit;
  19. public class JoinDemo3 {
  20.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         DataStream<Tuple2<String, Integer>> textStream = env.socketTextStream("localhost", 9000, "\n")
  24.                 .map(p -> {
  25.                     //输入格式为：user,1000,分别是用户名称和城市编号
  26.                     String[] list = p.split(",");
  27.                     return new Tuple2<String, Integer>(list[0], Integer.valueOf(list[1]));
  28.                 })
  29.                 .returns(new TypeHint<Tuple2<String, Integer>>() {
  30.                 });
  33.         DataStream<Tuple3<String,Integer, String>> orderedResult = AsyncDataStream
  34.                 //保证顺序：异步返回的结果保证顺序，超时时间1秒，最大容量2，超出容量触发反压
  35.                 .orderedWait(textStream, new JoinDemo3AyncFunction(), 1000L, TimeUnit.MILLISECONDS, 2)
  36.                 .setParallelism(1);
  38.         DataStream<Tuple3<String,Integer, String>> unorderedResult = AsyncDataStream
  39.                 //允许乱序：异步返回的结果允许乱序，超时时间1秒，最大容量2，超出容量触发反压
  40.                 .unorderedWait(textStream, new JoinDemo3AyncFunction(), 1000L, TimeUnit.MILLISECONDS, 2)
  41.                 .setParallelism(1);
  43.         orderedResult.print();
  44.         unorderedResult.print();
  45.         env.execute("joinDemo");
  46.     }
  48.     //定义个类，继承RichAsyncFunction，实现异步查询存储在mysql里的维表
  49.     //输入用户名、城市ID，返回 Tuple3<用户名、城市ID，城市名称>
  50.     static class JoinDemo3AyncFunction extends RichAsyncFunction<Tuple2<String, Integer>, Tuple3<String, Integer, String>> {
  51.         // 链接
  52.         private static String jdbcUrl = "jdbc:mysql://192.168.145.1:3306?useSSL=false";
  53.         private static String username = "root";
  54.         private static String password = "123";
  55.         private static String driverName = "com.mysql.jdbc.Driver";
  56.         java.sql.Connection conn;
  57.         PreparedStatement ps;
  59.         @Override
  60.         public void open(Configuration parameters) throws Exception {
  61.             super.open(parameters);
  63.             Class.forName(driverName);
  64.             conn = DriverManager.getConnection(jdbcUrl, username, password);
  65.             ps = conn.prepareStatement("select city_name from tmp.city_info where id = ?");
  66.         }
  68.         @Override
  69.         public void close() throws Exception {
  70.             super.close();
  71.             conn.close();
  72.         }
  74.         //异步查询方法
  75.         @Override
  76.         public void asyncInvoke(Tuple2<String, Integer> input, ResultFuture<Tuple3<String,Integer, String>> resultFuture) throws Exception {
  77.             // 使用 city id 查询
  78.             ps.setInt(1, input.f1);
  79.             ResultSet rs = ps.executeQuery();
  80.             String cityName = null;
  81.             if (rs.next()) {
  82.                 cityName = rs.getString(1);
  83.             }
  84.             List list = new ArrayList<Tuple2<Integer, String>>();
  85.             list.add(new Tuple3<>(input.f0,input.f1, cityName));
  86.             resultFuture.complete(list);
  87.         }
  89.         //超时处理
  90.         @Override
  91.         public void timeout(Tuple2<String, Integer> input, ResultFuture<Tuple3<String,Integer, String>> resultFuture) throws Exception {
  92.             List list = new ArrayList<Tuple2<Integer, String>>();
  93.             list.add(new Tuple3<>(input.f0,input.f1, ""));
  94.             resultFuture.complete(list);
  95.         }
  96.     }
  97. }

3. 广播维表

利用Flink的Broadcast State将维度数据流广播到下游做join操作。

优点：维度数据变更后可以即时更新到结果中。 缺点：数据保存在内存中，支持的维度数据量比较小。

示例：

  1. package join;
  3. import org.apache.flink.api.common.functions.RichMapFunction;
  4. import org.apache.flink.api.common.state.BroadcastState;
  5. import org.apache.flink.api.common.state.MapStateDescriptor;
  6. import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
  7. import org.apache.flink.api.common.typeinfo.TypeHint;
  8. import org.apache.flink.api.java.tuple.Tuple2;
  9. import org.apache.flink.api.java.tuple.Tuple3;
  10. import org.apache.flink.configuration.Configuration;
  11. import org.apache.flink.streaming.api.datastream.BroadcastStream;
  12. import org.apache.flink.streaming.api.datastream.DataStream;
  13. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  14. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
  15. import org.apache.flink.util.Collector;
  17. import java.util.ArrayList;
  18. import java.util.HashMap;
  19. import java.util.List;
  20. import java.util.Map;
  22. /**
  23.  * 这个例子是从socket中读取的流，数据为用户名称和城市id，维表是城市id、城市名称，
  24.  * 主流和维表关联，得到用户名称、城市id、城市名称
  25.  * 这个例子采用 Flink 广播流的方式来做为维度
  26.  **/
  27. public class JoinDemo4 {
  29.     public static void main(String[] args) throws Exception {
  30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  31.         //定义主流
  32.         DataStream<Tuple2<String, Integer>> textStream = env.socketTextStream("localhost", 9000, "\n")
  33.                 .map(p -> {
  34.                     //输入格式为：user,1000,分别是用户名称和城市编号
  35.                     String[] list = p.split(",");
  36.                     return new Tuple2<String, Integer>(list[0], Integer.valueOf(list[1]));
  37.                 })
  38.                 .returns(new TypeHint<Tuple2<String, Integer>>() {
  39.                 });
  41.         //定义城市流
  42.         DataStream<Tuple2<Integer, String>> cityStream = env.socketTextStream("localhost", 9001, "\n")
  43.                 .map(p -> {
  44.                     //输入格式为：城市ID,城市名称
  45.                     String[] list = p.split(",");
  46.                     return new Tuple2<Integer, String>(Integer.valueOf(list[0]), list[1]);
  47.                 })
  48.                 .returns(new TypeHint<Tuple2<Integer, String>>() {
  49.                 });
  51.         //将城市流定义为广播流
  52.         final MapStateDescriptor<Integer, String> broadcastDesc = new MapStateDescriptor("broad1", Integer.class, String.class);
  53.         BroadcastStream<Tuple2<Integer, String>> broadcastStream = cityStream.broadcast(broadcastDesc);
  55.         DataStream result = textStream.connect(broadcastStream)
  56.                 .process(new BroadcastProcessFunction<Tuple2<String, Integer>, Tuple2<Integer, String>, Tuple3<String, Integer, String>>() {
  57.                     //处理非广播流，关联维度
  58.                     @Override
  59.                     public void processElement(Tuple2<String, Integer> value, ReadOnlyContext ctx, Collector<Tuple3<String, Integer, String>> out) throws Exception {
  60.                         ReadOnlyBroadcastState<Integer, String> state = ctx.getBroadcastState(broadcastDesc);
  61.                         String cityName = "";
  62.                         if (state.contains(value.f1)) {
  63.                             cityName = state.get(value.f1);
  64.                         }
  65.                         out.collect(new Tuple3<>(value.f0, value.f1, cityName));
  66.                     }
  68.                     @Override
  69.                     public void processBroadcastElement(Tuple2<Integer, String> value, Context ctx, Collector<Tuple3<String, Integer, String>> out) throws Exception {
  70.                         System.out.println("收到广播数据：" + value);
  71.                         ctx.getBroadcastState(broadcastDesc).put(value.f0, value.f1);
  72.                     }
  73.                 });
  76.         result.print();
  77.         env.execute("joinDemo");
  78.     }
  79. }

4. Temporal table function join

Temporal table是持续变化表上某一时刻的视图，Temporal table function是一个表函数，传递一个时间参数，返回Temporal table这一指定时刻的视图。可以将维度数据流映射为Temporal table，主流与这个Temporal table进行关联，可以关联到某一个版本（历史上某一个时刻）的维度数据。

优点：维度数据量可以很大，维度数据更新及时，不依赖外部存储，可以关联不同版本的维度数据。 缺点：只支持在Flink SQL API中使用。

4.1 ProcessingTime示例
  1. package join;
  3. import org.apache.flink.api.common.typeinfo.TypeHint;
  4. import org.apache.flink.api.java.tuple.Tuple2;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.table.api.EnvironmentSettings;
  8. import org.apache.flink.table.api.Table;
  9. import org.apache.flink.table.api.java.StreamTableEnvironment;
  10. import org.apache.flink.table.functions.TemporalTableFunction;
  11. import org.apache.flink.types.Row;
  13. public class JoinDemo5 {
  14.     public static void main(String[] args) throws Exception {
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  17.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, bsSettings);
  19.         //定义主流
  20.         DataStream<Tuple2<String, Integer>> textStream = env.socketTextStream("localhost", 9000, "\n")
  21.                 .map(p -> {
  22.                     //输入格式为：user,1000,分别是用户名称和城市编号
  23.                     String[] list = p.split(",");
  24.                     return new Tuple2<String, Integer>(list[0], Integer.valueOf(list[1]));
  25.                 })
  26.                 .returns(new TypeHint<Tuple2<String, Integer>>() {
  27.                 });
  29.         //定义城市流
  30.         DataStream<Tuple2<Integer, String>> cityStream = env.socketTextStream("localhost", 9001, "\n")
  31.                 .map(p -> {
  32.                     //输入格式为：城市ID,城市名称
  33.                     String[] list = p.split(",");
  34.                     return new Tuple2<Integer, String>(Integer.valueOf(list[0]), list[1]);
  35.                 })
  36.                 .returns(new TypeHint<Tuple2<Integer, String>>() {
  37.                 });
  39.         //转变为Table
  40.         Table userTable = tableEnv.fromDataStream(textStream, "user_name,city_id,ps.proctime");
  41.         Table cityTable = tableEnv.fromDataStream(cityStream, "city_id,city_name,ps.proctime");
  43.         //定义一个TemporalTableFunction
  44.         TemporalTableFunction dimCity = cityTable.createTemporalTableFunction("ps", "city_id");
  45.         //注册表函数
  46.         tableEnv.registerFunction("dimCity", dimCity);
  48.         //关联查询
  49.         Table result = tableEnv
  50.                 .sqlQuery("select u.user_name,u.city_id,d.city_name from " + userTable + " as u " +
  51.                         ", Lateral table (dimCity(u.ps)) d " +
  52.                         "where u.city_id=d.city_id");
  54.         //打印输出
  55.         DataStream resultDs = tableEnv.toAppendStream(result, Row.class);
  56.         resultDs.print();
  57.         env.execute("joinDemo");
  58.     }
  59. }

4.2 EventTime示例
  1. package join;
  3. import org.apache.flink.api.java.tuple.Tuple3;
  4. import org.apache.flink.streaming.api.TimeCharacteristic;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
  8. import org.apache.flink.streaming.api.windowing.time.Time;
  9. import org.apache.flink.table.api.EnvironmentSettings;
  10. import org.apache.flink.table.api.Table;
  11. import org.apache.flink.table.api.java.StreamTableEnvironment;
  12. import org.apache.flink.table.functions.TemporalTableFunction;
  13. import org.apache.flink.types.Row;
  15. import java.sql.Timestamp;
  16. import java.util.ArrayList;
  17. import java.util.List;
  19. public class JoinDemo9 {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         //指定是EventTime
  23.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  24.         EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  25.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, bsSettings);
  26.         env.setParallelism(1);
  28.         //主流，用户流, 格式为：user_name、city_id、ts
  29.         List<Tuple3<String, Integer, Long>> list1 = new ArrayList<>();
  30.         list1.add(new Tuple3<>("user1", 1001, 1L));
  31.         list1.add(new Tuple3<>("user1", 1001, 10L));
  32.         list1.add(new Tuple3<>("user2", 1002, 2L));
  33.         list1.add(new Tuple3<>("user2", 1002, 15L));
  34.         DataStream<Tuple3<String, Integer, Long>> textStream = env.fromCollection(list1)
  35.                 .assignTimestampsAndWatermarks(
  36.                         //指定水位线、时间戳
  37.                         new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Integer, Long>>(Time.seconds(10)) {
  38.                             @Override
  39.                             public long extractTimestamp(Tuple3<String, Integer, Long> element) {
  40.                                 return element.f2;
  41.                             }
  42.                         }
  43.                 );
  45.         //定义城市流,格式为：city_id、city_name、ts
  46.         List<Tuple3<Integer, String, Long>> list2 = new ArrayList<>();
  47.         list2.add(new Tuple3<>(1001, "beijing", 1L));
  48.         list2.add(new Tuple3<>(1001, "beijing2", 10L));
  49.         list2.add(new Tuple3<>(1002, "shanghai", 1L));
  50.         list2.add(new Tuple3<>(1002, "shanghai2", 5L));
  52.         DataStream<Tuple3<Integer, String, Long>> cityStream = env.fromCollection(list2)
  53.                 .assignTimestampsAndWatermarks(
  54.                         //指定水位线、时间戳
  55.                         new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Integer, String, Long>>(Time.seconds(10)) {
  56.                             @Override
  57.                             public long extractTimestamp(Tuple3<Integer, String, Long> element) {
  58.                                 return element.f2;
  59.                             }
  60.                         });
  62.         //转变为Table
  63.         Table userTable = tableEnv.fromDataStream(textStream, "user_name,city_id,ts.rowtime");
  64.         Table cityTable = tableEnv.fromDataStream(cityStream, "city_id,city_name,ts.rowtime");
  66.         tableEnv.createTemporaryView("userTable", userTable);
  67.         tableEnv.createTemporaryView("cityTable", cityTable);
  69.         //定义一个TemporalTableFunction
  70.         TemporalTableFunction dimCity = cityTable.createTemporalTableFunction("ts", "city_id");
  71.         //注册表函数
  72.         tableEnv.registerFunction("dimCity", dimCity);
  74.         //关联查询
  75.         Table result = tableEnv
  76.                 .sqlQuery("select u.user_name,u.city_id,d.city_name,u.ts from userTable as u " +
  77.                         ", Lateral table (dimCity(u.ts)) d " +
  78.                         "where u.city_id=d.city_id");
  80.         //打印输出
  81.         DataStream resultDs = tableEnv.toAppendStream(result, Row.class);
  82.         resultDs.print();
  83.         env.execute("joinDemo");
  84.     }
  85. }

结果输出为：

  1. user1,1001,beijing,1970-01-01T00:00:00.001
  2. user1,1001,beijing2,1970-01-01T00:00:00.010
  3. user2,1002,shanghai,1970-01-01T00:00:00.002
  4. user2,1002,shanghai2,1970-01-01T00:00:00.015

通过结果可以看到，根据主流中的EventTime的时间，去维表流中取响应时间版本的数据。

4.3 Kafka Source的EventTime实例
  1. package join.temporaltablefunctionjoin;
  3. import lombok.Data;
  4. import org.apache.flink.streaming.api.TimeCharacteristic;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
  8. import org.apache.flink.streaming.api.windowing.time.Time;
  9. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  10. import org.apache.flink.table.api.EnvironmentSettings;
  11. import org.apache.flink.table.api.Table;
  12. import org.apache.flink.table.api.java.StreamTableEnvironment;
  13. import org.apache.flink.table.functions.TemporalTableFunction;
  14. import org.apache.flink.types.Row;
  16. import java.io.Serializable;
  17. import java.util.Properties;
  19. public class JoinDemo10 {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         //指定是EventTime
  23.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  24.         EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  25.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, bsSettings);
  26.         env.setParallelism(1);
  28.         //Kafka的ip和要消费的topic,//Kafka设置
  29.         String kafkaIPs = "192.168.***.**1:9092,192.168.***.**2:9092,192.168.***.**3:9092";
  30.         Properties props = new Properties();
  31.         props.setProperty("bootstrap.servers", kafkaIPs);
  32.         props.setProperty("group.id", "group.cyb.2");
  34.         //读取用户信息Kafka
  35.         FlinkKafkaConsumer<UserInfo> userConsumer = new FlinkKafkaConsumer<UserInfo>("user", new UserInfoSchema(), props);
  36.         userConsumer.setStartFromEarliest();
  37.         userConsumer.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<UserInfo>(Time.seconds(0)) {
  38.             @Override
  39.             public long extractTimestamp(UserInfo userInfo) {
  40.                 return userInfo.getTs();
  41.             }
  42.         });
  44.         //读取城市维度信息Kafka
  45.         FlinkKafkaConsumer<CityInfo> cityConsumer = new FlinkKafkaConsumer<CityInfo>("city", new CityInfoSchema(), props);
  46.         cityConsumer.setStartFromEarliest();
  47.         cityConsumer.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<CityInfo>(Time.seconds(0)) {
  48.             @Override
  49.             public long extractTimestamp(CityInfo cityInfo) {
  50.                 return cityInfo.getTs();
  51.             }
  52.         });
  54.         //主流，用户流, 格式为：user_name、city_id、ts
  55.         Table userTable = tableEnv.fromDataStream(env.addSource(userConsumer),"userName,cityId,ts.rowtime" );
  56.         //定义城市维度流,格式为：city_id、city_name、ts
  57.         Table cityTable = tableEnv.fromDataStream(env.addSource(cityConsumer),"cityId,cityName,ts.rowtime");
  58.         tableEnv.createTemporaryView("userTable", userTable);
  59.         tableEnv.createTemporaryView("cityTable", cityTable);
  61.         //定义一个TemporalTableFunction
  62.         TemporalTableFunction dimCity = cityTable.createTemporalTableFunction("ts", "cityId");
  63.         //注册表函数
  64.         tableEnv.registerFunction("dimCity", dimCity);
  66.         Table u = tableEnv.sqlQuery("select * from userTable");
  67.         u.printSchema();
  68.         tableEnv.toAppendStream(u, Row.class).print("用户流接收到：");
  70.         Table c = tableEnv.sqlQuery("select * from cityTable");
  71.         c.printSchema();
  72.         tableEnv.toAppendStream(c, Row.class).print("城市流接收到：");
  74.         //关联查询
  75.         Table result = tableEnv
  76.                 .sqlQuery("select u.userName,u.cityId,d.cityName,u.ts " +
  77.                         "from userTable as u " +
  78.                         ", Lateral table  (dimCity(u.ts)) d " +
  79.                         "where u.cityId=d.cityId");
  81.         //打印输出
  82.         DataStream resultDs = tableEnv.toAppendStream(result, Row.class);
  83.         resultDs.print("\t\t关联输出：");
  84.         env.execute("joinDemo");
  85.     }
  86. }
  1. package join.temporaltablefunctionjoin;
  2. import java.io.Serializable;
  4.  @Data
  5. public class UserInfo implements Serializable {
  6.     private String userName;
  7.     private Integer cityId;
  8.     private Long ts;
  9. }
  1. package join.temporaltablefunctionjoin;
  2. import java.io.Serializable;
  4. @Data
  5. public class CityInfo implements Serializable {
  6.     private Integer cityId;
  7.     private String cityName;
  8.     private Long ts;
  9. }
  1. package join.temporaltablefunctionjoin;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.TypeReference;
  5. import org.apache.flink.api.common.typeinfo.TypeHint;
  6. import org.apache.flink.api.common.typeinfo.TypeInformation;
  7. import org.apache.flink.api.common.serialization.DeserializationSchema;
  9. import java.io.IOException;
  10. import java.nio.charset.StandardCharsets;
  12. public class UserInfoSchema implements DeserializationSchema<UserInfo> {
  14.     @Override
  15.     public UserInfo deserialize(byte[] message) throws IOException {
  16.         String jsonStr = new String(message, StandardCharsets.UTF_8);
  17.         UserInfo data = JSON.parseObject(jsonStr, new TypeReference<UserInfo>() {});
  18.         return data;
  19.     }
  21.     @Override
  22.     public boolean isEndOfStream(UserInfo nextElement) {
  23.         return false;
  24.     }
  26.     @Override
  27.     public TypeInformation<UserInfo> getProducedType() {
  28.         return TypeInformation.of(new TypeHint<UserInfo>() {
  29.         });
  30.     }
  31. }
  1. package join.temporaltablefunctionjoin;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.TypeReference;
  5. import org.apache.flink.api.common.serialization.DeserializationSchema;
  6. import org.apache.flink.api.common.typeinfo.TypeHint;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  9. import java.io.IOException;
  10. import java.nio.charset.StandardCharsets;
  12. public class CityInfoSchema implements DeserializationSchema<CityInfo> {
  14.     @Override
  15.     public CityInfo deserialize(byte[] message) throws IOException {
  16.         String jsonStr = new String(message, StandardCharsets.UTF_8);
  17.         CityInfo data = JSON.parseObject(jsonStr, new TypeReference<CityInfo>() {});
  18.         return data;
  19.     }
  21.     @Override
  22.     public boolean isEndOfStream(CityInfo nextElement) {
  23.         return false;
  24.     }
  26.     @Override
  27.     public TypeInformation<CityInfo> getProducedType() {
  28.         return TypeInformation.of(new TypeHint<CityInfo>() {
  29.         });
  30.     }
  31. }

依次向user和city两个topic中写入数据:

  1. 用户信息格式：{“userName”:“user1”,“cityId”:1,“ts”:11}
  2. 城市维度格式：{“cityId”:1,“cityName”:“nanjing”,“ts”:15}

测试得到的输出如下：

  1. 城市流接收到：> 1,beijing,1970-01-01T00:00
  2. 用户流接收到：> user1,1,1970-01-01T00:00
  3.         关联输出：> user1,1,beijing,1970-01-01T00:00
  4. 城市流接收到：> 1,shanghai,1970-01-01T00:00:00.005
  5. 用户流接收到：> user1,1,1970-01-01T00:00:00.001
  6.         关联输出：> user1,1,beijing,1970-01-01T00:00:00.001
  7. 用户流接收到：> user1,1,1970-01-01T00:00:00.004
  8.         关联输出：> user1,1,beijing,1970-01-01T00:00:00.004
  9. 用户流接收到：> user1,1,1970-01-01T00:00:00.005
  10.         关联输出：> user1,1,shanghai,1970-01-01T00:00:00.005
  11. 用户流接收到：> user1,1,1970-01-01T00:00:00.007
  12. 用户流接收到：> user1,1,1970-01-01T00:00:00.009
  13. 城市流接收到：> 1,shanghai,1970-01-01T00:00:00.007
  14.         关联输出：> user1,1,shanghai,1970-01-01T00:00:00.007
  15. 城市流接收到：> 1,wuhan,1970-01-01T00:00:00.010
  16.         关联输出：> user1,1,shanghai,1970-01-01T00:00:00.009
  17. 用户流接收到：> user1,1,1970-01-01T00:00:00.011
  18. 城市流接收到：> 1,nanjing,1970-01-01T00:00:00.015
  19.         关联输出：> user1,1,wuhan,1970-01-01T00:00:00.011

5. 四种维表关联方式总结

Flink维表Join实践 - 图4