Flink核心API

客官°小女子只卖身不卖艺 2022-09-09 09:23 221阅读 0赞

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16]  
Flink中提供了4种不同层次的API，每种API在简洁和易表达之间有自己的权衡，适用于不同的场景。目前前3个会用得比较多。

*  **SQL**:Flink的SQL底层是基于Apache Calcite，Apache Calcite实现了标准的SQL，使用起来比其他API更加灵活，因为可以直接使用SQL语句。Table API和SQL可以很容易地结合在一块使用，因为它们都返回Table对象。
 *  **Table API**:一般与DataSet或者DataStream紧密关联，可以通过一个DataSet或DataStream创建出一个Table，然后再使用类似于filter, join,或者select这种操作。最后还可以将一个Table对象转成 DataSet或DataStream。
 *  **核心API(DataStream/DataSet API)**:主要提供了针对流数据和批数据的处理，是对低级API进行了一 些封装，提供了filter、sum、max、min等高级函数，简单易用，所以这些API在工作中应用还是比较广泛的。
 *  **低级API(Stateful Stream Processing)**:提供了对时间和状态的细粒度控制，简洁性和易用性较差， 主要应用在一些复杂事件处理逻辑上。

针对这些api，我们主要讲解以下内容：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 1]

### DataStream API ###

DataStream API主要分为3块: `DataSource`、`Transformation`、`DataSink`。

DataSource是程序的输入数据源。  
Transformation是具体的操作，它对一个或多个输入数据源进行计算处理，例如map、flatMap和filter等操作。  
DataSink是程序的输出，它可以把Transformation处理之后的数据输出到指定的存储介质中。

#### DataStream API之DataSoure ####

DataSource是程序的输入数据源，Flink提供了大量内置的DataSource，也支持自定义DataSource，不过目前Flink提供的这些已经足够我们正常使用了。

Flink提供的内置输入数据源:包括基于`socket`、基于`Collection`  
还有就是Flink还提供了一批`Connectors`，可以实现读取第三方数据源

Flink 内置:表示Flink中默认自带的。  
Apache Bahir:表示需要添加这个依赖包之后才能使用的

<table> 
 <thead> 
  <tr> 
   <th>Flink内置</th> 
   <th>Apache Bahir</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Kafka</td> 
   <td>ActiveMQ</td> 
  </tr> 
  <tr> 
   <td>Kinesis Streams</td> 
   <td>Netty</td> 
  </tr> 
  <tr> 
   <td>RabbitMQ</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>NIFI</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Twitter Streaming API</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Google PubSub</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

针对source的这些Connector，我们在实际工作中最常用的就是Kafka

当程序出现错误的时候，Flink的容错机制能恢复并继续运行程序，这种错误包括机器故障、网络故障、程序故障等

针对Flink提供的常用数据源接口，如果程序开启了checkpoint快照机制，Flink可以提供这些容错性保证

<table> 
 <thead> 
  <tr> 
   <th>DataSource</th> 
   <th>容错包证</th> 
   <th></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Socket</td> 
   <td>at most once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Collection</td> 
   <td>exactly once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Kafka</td> 
   <td>exactly once</td> 
   <td>需要使用0.10及以上版本</td> 
  </tr> 
 </tbody> 
</table>

#### DataStream API之Transformation ####

transformation是Flink程序的计算算子，负责对数据进行处理，Flink提供了大量的算子，其实Flink中的大部分算子的使用和spark中算子的使用是一样的，下面我们来看一下:

<table> 
 <thead> 
  <tr> 
   <th>算子</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>map</td> 
   <td>输入一个元素进行处理，返回一个元素</td> 
  </tr> 
  <tr> 
   <td>flatMap</td> 
   <td>输入一个元素进行处理，可以返回多个元素</td> 
  </tr> 
  <tr> 
   <td>filter</td> 
   <td>对数据进行过滤，符合条件的数据会被留下</td> 
  </tr> 
  <tr> 
   <td>keyBy</td> 
   <td>根据key分组，相同key的数据会进入同一个分区</td> 
  </tr> 
  <tr> 
   <td>reduce</td> 
   <td>对当前元素和上一次的结果进行聚合操作</td> 
  </tr> 
  <tr> 
   <td>aggregations</td> 
   <td>sum(),min(),max()等</td> 
  </tr> 
 </tbody> 
</table>

这里的算子的用法和spark中对应算子的用法基本一致  
另外也有一些Flink独有的算子

<table> 
 <thead> 
  <tr> 
   <th>算子</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>union</td> 
   <td>合并多个流，多个流的数据类型必须一致</td> 
  </tr> 
  <tr> 
   <td>connect</td> 
   <td>只能连接两个流，两个流的数据类型可以不同</td> 
  </tr> 
  <tr> 
   <td>split</td> 
   <td>根据规则把一个数据流切分为多个流</td> 
  </tr> 
  <tr> 
   <td>shuffle</td> 
   <td>随机分区</td> 
  </tr> 
  <tr> 
   <td>rebalance</td> 
   <td>对数据集进行再平衡，重分区，消除数据倾斜</td> 
  </tr> 
  <tr> 
   <td>rescale</td> 
   <td>重分区</td> 
  </tr> 
  <tr> 
   <td>partitionCustom</td> 
   <td>自定义分区</td> 
  </tr> 
 </tbody> 
</table>

*  **union**: 表示合并多个流，但是多个流的数据类型必须一致

多个流join之后，就变成了一个流  
应用场景:多种数据源的数据类型一致，数据处理规则也一致

演示代码:  
**Scala版本**

object StreamUnionScala { 
      	def main(args: Array[String]): Unit = { 
    	    val env = StreamExecutionEnvironment.getExecutionEnvironment
    	
    	    import org.apache.flink.api.scala._
    	    //第1份数据流
    	    val text1 = env.fromCollection(Array(1, 2, 3, 4, 5))
    	    //第2份数据流
    	    val text2 = env.fromCollection(Array(6, 7, 8, 9, 10))
    	
    	    //合并流
    	    val unionStream = text1.union(text2)
    	
    	    //打印流中的数据
    	    unionStream.print().setParallelism(1)
    	
    	    env.execute("StreamUnionScala")
      	}
    }

**Java版本**

public class StreamUnionJava { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //第1份数据流
            DataStreamSource<Integer> text1 = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5));
            //第2份数据流
            DataStreamSource<Integer> text2 = env.fromCollection(Arrays.asList(6, 7, 8, 9, 10));
    
            //合并流
            DataStream<Integer> unionStream = text1.union(text2);
    
            //打印流中的数据
            unionStream.print().setParallelism(1);
    
            env.execute("StreamUnionJava");
        }
    }

*  **connect**:只能连接两个流，两个流的数据类型可以不同

两个流被connect之后，只是被放到了同一个流中，它们内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。  
connect方法会返回connectedStream ， 在connectedStream中需要使用CoMap 、CoFlatMap这种函数，类似于map和flatmap

演示代码:  
**Scala版本**

object StreamConnectScala { 
      	def main(args: Array[String]): Unit = { 
    	    val env = StreamExecutionEnvironment.getExecutionEnvironment
    	
    	    import org.apache.flink.api.scala._
    	    //第1份数据流
    	    val text1 = env.fromElements("user:tom,age:18")
    	    //第2份数据流
    	    val text2 = env.fromElements("user:jack_age:20")
    	
    	    //连接两个流
    	    val connectStream = text1.connect(text2)
    	
    	    connectStream.map(new CoMapFunction[String,String,String] { 
    	      //处理第1份数据流中的数据
    	      override def map1(value: String): String = { 
    	        value.replace(",","-")
    	      }
    	      //处理第2份数据流中的数据
    	      override def map2(value: String): String = { 
    	        value.replace("_","-")
    	      }
    	    }).print().setParallelism(1)
    	
    	    env.execute("StreamConnectScala")
      	}
    }

**Java版本**

public class StreamConnectJava { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //第1份数据流
            DataStreamSource<String> text1 = env.fromElements("user:tom,age:18");
            //第2份数据流
            DataStreamSource<String> text2 = env.fromElements("user:jack_age:18");
    
            //连接两个流
            ConnectedStreams<String, String> connectStream = text1.connect(text2);
    
            connectStream.map(new CoMapFunction<String, String, String>() { 
    
                //处理第1份数据流中的数据
                @Override
                public String map1(String value) throws Exception { 
                    return value.replace(",","-");
                }
                //处理第2份数据流中的数据
                @Override
                public String map2(String value) throws Exception { 
                    return value.replace("_","-");
                }
            }).print().setParallelism(1);
    
            env.execute("StreamConnectJava");
        }
    }

*  **split**:根据规则把一个数据流切分为多个流

> 注意:split只能分一次流，切分出来的流不能继续分流

split需要和select配合使用，选择切分后的流  
应用场景:将一份数据流切分为多份，便于针对每一份数据使用不同的处理逻辑

演示代码:  
**Scala版本**

object StreamSplitScala { 
      	def main(args: Array[String]): Unit = { 
    	    val env = StreamExecutionEnvironment.getExecutionEnvironment
    	
    	    import org.apache.flink.api.scala._
    	    //初始化数据
    	    val text = env.fromCollection(Array(1, 2, 4, 5, 6, 7, 8, 9, 10))
    	
    	    //按照数据的奇偶性对数据进行分流
    	    val splitStream = text.split(new OutputSelector[Int] { 
    	      override def select(value: Int): lang.Iterable[String] = { 
    	        val list = new util.ArrayList[String]()
    	        if(value % 2 == 0){ 
    	          list.add("even")//偶数
    	        }else{ 
    	          list.add("odd")//奇数
    	        }
    	        list
    	      }
    	    })
    	
    	    //选择流
    	    val evenStream = splitStream.select("even")
    	    evenStream.print().setParallelism(1)
    	
    	    //二次切流会报错
    	    //Consecutive multiple splits are not supported. Splits are deprecated. Please use side-outputs.
    	    /*val lowHighStream = evenStream.split(new OutputSelector[Int] { override def select(value: Int): lang.Iterable[String] = { val list = new util.ArrayList[String]() if(value <= 5){ list.add("low") }else{ list.add("high") } list } }) val lowStream = lowHighStream.select("low") lowStream.print().setParallelism(1)*/
    	
    	
    	    env.execute("StreamSplitScala")
      	}
    }

**Java版本**

public class StreamSplitJava { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStreamSource<Integer> text = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
    
            //按照数据的奇偶性对数据进行分流
            SplitStream<Integer> splitStream = text.split(new OutputSelector<Integer>() { 
                @Override
                public Iterable<String> select(Integer value) { 
                    ArrayList<String> list = new ArrayList<>();
                    if (value % 2 == 0) { 
                        list.add("even");//偶数
                    } else { 
                        list.add("odd");//奇数
                    }
                    return list;
                }
            });
    
            //选择流
            DataStream<Integer> enevStream = splitStream.select("even");
    
            enevStream.print().setParallelism(1);
    
            env.execute("StreamSplitJava");
        }
    }

目前split切分的流无法进行二次切分，并且split方法已经标记为过时了，官方不推荐使用，现在官方推荐使用`side output`的方式实现。

下面我来看一下使用side output如何实现流的多次切分  
**Scala版本**

object StreamSideOutputScala { 
      	def main(args: Array[String]): Unit = { 
    	    val env = StreamExecutionEnvironment.getExecutionEnvironment
    	
    	    import org.apache.flink.api.scala._
    	    val text = env.fromCollection(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    	
    	    //按照数据的奇偶性对数据进行分流
    	    //首先定义两个sideoutput来准备保存切分出来的数据
    	    val outputTag1 = new OutputTag[Int]("even")//保存偶数
    	    val outputTag2 = new OutputTag[Int]("odd")//保存奇数
    	
    	    //注意：process属于Flink中的低级api
    	    val outputStream = text.process(new ProcessFunction[Int, Int] { 
    	      override def processElement(value: Int, ctx: ProcessFunction[Int, Int]#Context, out: Collector[Int]): Unit = { 
    	        if (value % 2 == 0) { 
    	          ctx.output(outputTag1, value)
    	        } else { 
    	          ctx.output(outputTag2, value)
    	        }
    	      }
    	    })
    	
    	    //获取偶数数据流
    	    val evenStream = outputStream.getSideOutput(outputTag1)
    	    //获取奇数数据流
    	    val oddStream = outputStream.getSideOutput(outputTag2)
    	    //evenStream.print().setParallelism(1)
    	
    	    //对evenStream流进行二次切分
    	    val outputTag11 = new OutputTag[Int]("low")//保存小于等于5的数字
    	    val outputTag12 = new OutputTag[Int]("high")//保存大于5的数字
    	
    	    val subOutputStream = evenStream.process(new ProcessFunction[Int, Int] { 
    	      override def processElement(value: Int, ctx: ProcessFunction[Int, Int]#Context, out: Collector[Int]): Unit = { 
    	        if (value <= 5) { 
    	          ctx.output(outputTag11, value)
    	        } else { 
    	          ctx.output(outputTag12, value)
    	        }
    	      }
    	    })
    	
    	    //获取小于等于5的数据流
    	    val lowStream = subOutputStream.getSideOutput(outputTag11)
    	    //获取大于5的数据流
    	    val highStream = subOutputStream.getSideOutput(outputTag12)
    	
    	    lowStream.print().setParallelism(1)
    	
    	    env.execute("StreamSideOutputScala")
    
      	}
    
    }

**Java版本**

public class StreamSideOutputJava { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStreamSource<Integer> text = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
    
            //按照数据的奇偶性对数据进行分流
            //首先定义两个sideoutput来保存切分出来的数据
            OutputTag<Integer> outputTag1 = new OutputTag<Integer>("even") { };//保存偶数
            OutputTag<Integer> outputTag2 = new OutputTag<Integer>("odd") { };//保存奇数
    
            SingleOutputStreamOperator<Integer> outputStream = text.process(new ProcessFunction<Integer, Integer>() { 
    
                @Override
                public void processElement(Integer value, Context ctx, Collector<Integer> out)
                        throws Exception { 
                    if (value % 2 == 0) { 
                        ctx.output(outputTag1, value);
                    } else { 
                        ctx.output(outputTag2, value);
                    }
                }
            });
    
            //获取偶数数据流
            DataStream<Integer> evenStream = outputStream.getSideOutput(outputTag1);
            //获取奇数数据流
            DataStream<Integer> oddStream = outputStream.getSideOutput(outputTag2);
    
            //对evenStream流进行二次切分
            OutputTag<Integer> outputTag11 = new OutputTag<Integer>("low") { };
            OutputTag<Integer> outputTag12 = new OutputTag<Integer>("high") { };
            SingleOutputStreamOperator<Integer> subOutputStream = evenStream.process(new ProcessFunction<Integer, Integer>() { 
    
                @Override
                public void processElement(Integer value, Context ctx, Collector<Integer> out)
                        throws Exception { 
                    if (value <= 5) { 
                        ctx.output(outputTag11, value);
                    } else { 
                        ctx.output(outputTag12, value);
                    }
                }
            });
    
            //获取小于等于5的数据流
            DataStream<Integer> lowStream = subOutputStream.getSideOutput(outputTag11);
            //获取大于5的数据流
            DataStream<Integer> highStream = subOutputStream.getSideOutput(outputTag12);
            lowStream.print().setParallelism(1);
    
            env.execute("StreamSideOutputJava");
        }
    }

其实想要实现多级流切分，使用`filter`算子也是可以实现的

最后我们针对几个算子总结一下：

首先是**union和connect的区别**，如图所示:  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
union可以连接多个流，最后汇总成一个流，流里面的数据使用相同的计算规则  
connect值可以连接2个流，最后汇总成一个流，但是流里面的两份数据相互还是独立的，每一份数据使用一个计算规则

然后是**流切分**  
如果是只需要切分一次的话使用`split`或者`side output`都可以  
如果想要切分多次，就不能使用`split`了，需要使用`side output`  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
接下来看一下这几个和分区相关的算子

<table> 
 <thead> 
  <tr> 
   <th>算子</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>random</td> 
   <td>随机分区</td> 
  </tr> 
  <tr> 
   <td>rebalance</td> 
   <td>对数据集进行再平衡，重分区，消除数据倾斜</td> 
  </tr> 
  <tr> 
   <td>rescale</td> 
   <td>重分区</td> 
  </tr> 
  <tr> 
   <td>broadcast</td> 
   <td>广播分区</td> 
  </tr> 
  <tr> 
   <td>custom partition</td> 
   <td>自定义分区</td> 
  </tr> 
 </tbody> 
</table>

**random**: 随机分区，它表示将上游数据随机分发到下游算子实例的每个分区中，在代码层面体现是调用`shuffle()`函数  
shuffle底层对应的是ShufflePartitioner这个类  
这个类里面有一个selectChannel函数，这个函数会计算数据将会被发送给哪个分区，里面使用的是 random.nextInt，所以说是随机的。

public int selectChannel(SerializationDelegate<StreamRecord<T>> record) { 
       return random.nextInt(numberOfChannels);
    }

**rebalance**: 重新平衡分区(循环分区)，我觉得叫循环分区更好理解，它表示对数据集进行再平衡，消除 数据倾斜，为每个分区创建相同的负载，其实就是通过循环的方式给下游算子实例的每个分区分配数据， 在代码层面体现是调用`rebalance()`函数  
rebalance底层对应的是`RebalancePartitioner`这个类  
这个类里面有一个`setup`和`selectChannel`函数，setup函数会根据分区数初始化一个随机值 `nextChannelToSendTo` ，然后selectChannel函数会使用nextChannelToSendTo 加1和分区数取模，把计算的值再赋给`nextChannelToSendTo` ，后面以此类推，其实就可以实现向下游算子实例的多个分区循环发送数据了，这样每个分区获取到的数据基本一致。

public void setup(int numberOfChannels) { 
       super.setup(numberOfChannels);
       nextChannelToSendTo = ThreadLocalRandom.current().nextInt(numberOfChannels
    }
    public int selectChannel(SerializationDelegate<StreamRecord<T>> record) { 
       nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels;
       return nextChannelToSendTo;
    }

**rescale**:重分区  
rescale底层对应的是`RescalePartitioner`这个类  
这个类里面有一个selectChannel函数，这里面的numberOfChannels是分区数量，其实也可以认为是我们所说的算子的并行度，因为一个分区是由一个线程负责处理的，它们两个是一一对应的。

public int selectChannel(SerializationDelegate<StreamRecord<T>> record) { 
       if (++nextChannelToSendTo >= numberOfChannels) { 
          nextChannelToSendTo = 0;
       }
       return nextChannelToSendTo;
    }

源码注释也有写清楚： 如果上游操作有2个并发，而下游操作有4个并发，那么上游的1个并发结果循环分配给下游的2个并发操 作，上游的另外1个并发结果循环分配给下游的另外2个并发操作。另一种情况，如果上游有4个并发操作 而下游有2个并发操作，那么上游的其中2个并发操作的结果会分配给下游的一个并发操作，而上游的另 外2个并发操作的结果则分配给下游的另外1个并发操作。

**broadcast**:广播分区，将上游算子实例中的数据输出到下游算子实例的每个分区中，适合用于大数据集join小数据集的场景，可以提高性能  
broadcast底层对应的是`BroadcastPartitione`r这个类  
看这个类中的`selectChannel`函数代码的注释，提示广播分区不支持选择Channel,因为会输出数据到下游的每个Channel中，就是发送到下游算子实例的每个分区中

/** * Note: Broadcast mode could be handled directly for all the output channels * in record writer, so it is no need to select channels via this method. */
    @Override
    public int selectChannel(SerializationDelegate<StreamRecord<T>> record) { 
       throw new UnsupportedOperationException("Broadcast partitioner does not su
    }

**custom partition**:自定义分区，可以按照自定义规则实现  
自定义分区需要实现`Partitioner`接口

#### DataStream API之DataSink ####

DataSink是 输出组件，负责把计算好的数据输出到其它存储介质中  
Flink支持把流数据输出到文件中，不过在实际工作中这种场景不多，因为流数据处理之后一般会存储到一些消息队列里面，或者数据库里面，很少会保存到文件中的。

还有就是print，直接打印，这种用法主要是在测试的时候使用的， 便查看输出的结果信息

Flink提供了一批Connectors，可以实现输出到第三方目的地

<table> 
 <thead> 
  <tr> 
   <th>Fink内置</th> 
   <th>Apache Bahir</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Kafka</td> 
   <td>ActiveMQ</td> 
  </tr> 
  <tr> 
   <td>Cassandra</td> 
   <td>Flume</td> 
  </tr> 
  <tr> 
   <td>Kinesis Streams</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Elasticsearch</td> 
   <td>Redis</td> 
  </tr> 
  <tr> 
   <td>Hadoop FileSysterm</td> 
   <td>Akka</td> 
  </tr> 
  <tr> 
   <td>RabbitMQ</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>NiFi</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>JDBC</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

针对sink的这些connector，我们在实际工作中最常用的是`kafka`、`redis`  
针对Flink提供的常用sink组件，可以提供这些容错性保证

<table> 
 <thead> 
  <tr> 
   <th>DataSink</th> 
   <th>容错保证</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Redis</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Kafka</td> 
   <td>at least once / exactly once</td> 
   <td>Kafka0.9和0.10提供at least once，Kafka0.11及以上提供 exactly once</td> 
  </tr> 
 </tbody> 
</table>

我们来看一下示例代码  
**Scala版本**

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.connectors.redis.RedisSink
    import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
    import org.apache.flink.streaming.connectors.redis.common.mapper.{ RedisCommand, RedisCommandDescription, RedisMapper}
    
    /** * 需求：接收socket传输过来的数据，把数据保存到redis的list队列中 */
    object StreamRedisSinkScala { 
      	def main(args: Array[String]): Unit = { 
    	    val env = StreamExecutionEnvironment.getExecutionEnvironment
    	
    	    //连接socket获取输入数据
    	    val text = env.socketTextStream("bigdata", 9001)
    	
    	    import org.apache.flink.api.scala._
    	    //组装数据，这里组装的是tuple2类型
    	    //第一个元素是指list队列的key名称
    	    //第二个元素是指需要向list队列中添加的元素
    	    val listData = text.map(word => ("l_words_scala", word))
    	
    	    //指定redissink
    	    val conf = new FlinkJedisPoolConfig.Builder().setHost("bigdata").setPort(6379).build()
    	    val redisSink = new RedisSink[Tuple2[String, String]](conf, new MyRedisMapper)
    	    listData.addSink(redisSink)
    	
    	    env.execute("StreamRedisSinkScala")
      	}
    
      	class MyRedisMapper extends RedisMapper[Tuple2[String,String]]{ 
    	    //指定具体的操作命令
    	    override def getCommandDescription: RedisCommandDescription = { 
    	      new RedisCommandDescription(RedisCommand.LPUSH)
    	    }
    	
    	    //获取key
    	    override def getKeyFromData(data: (String, String)): String = { 
    	      data._1
    	    }
    	    //获取value
    	    override def getValueFromData(data: (String, String)): String = { 
    	      data._2
    	    }
      	}
    
    }

**Java版本**

public class StreamRedisSinkJava { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //连接socket获取输入数据
            DataStreamSource<String> text = env.socketTextStream("bigdata", 9001);
    
            //组装数据
            SingleOutputStreamOperator<Tuple2<String, String>> listData = text.map(new MapFunction<String, Tuple2<String, String>>() { 
                @Override
                public Tuple2<String, String> map(String word) throws Exception { 
                    return new Tuple2<String, String>("l_words_java", word);
                }
            });
    
            //指定redissink
            FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("bigdata").setPort(6379).build();
            RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(conf, new MyRedisMapper());
            listData.addSink(redisSink);
    
            env.execute("StreamRedisSinkJava");
        }
    
        public static class MyRedisMapper implements RedisMapper<Tuple2<String,String>>{ 
    
            //指定具体的操作命令
            @Override
            public RedisCommandDescription getCommandDescription() { 
                return new RedisCommandDescription(RedisCommand.LPUSH);
            }
    
            //获取key
            @Override
            public String getKeyFromData(Tuple2<String, String> data) { 
                return data.f0;
            }
            //获取value
            @Override
            public String getValueFromData(Tuple2<String, String> data) { 
                return data.f1;
            }
        }
    }

注 意 :  
针对List数据类型 ，我们在定义`getCommandDescription`方法的时候，使用`new RedisCommandDescription(RedisCommand.LPUSH);`。  
如果是Hash数据类型 ，在定义`getCommandDescription`方法的时候 ，需要使用`new RedisCommandDescription(RedisCommand.HSET,“hashKey”);`，在构造函数中需要直接指定Hash数据类型的key的名称。

### Table API & SQL ###

如果想要使用Table API 和SQL的话，需要添加下面的依赖

<dependency>
      	<groupId>org.apache.flink</groupId>
    	<artifactId>flink-table-api-java-bridge_2.12</artifactId>
    	<version>1.11.0</version>
    	<scope>provided</scope>
    </dependency>
    <dependency>
    	<groupId>org.apache.flink</groupId>
    	<artifactId>flink-table-api-scala-bridge_2.12</artifactId>
    	<version>1.11.0</version>
    	<scope>provided</scope>
    </dependency>

如果想在本地IDE中运行程序，还需要添加下面的依赖

<dependency>
    	<groupId>org.apache.flink</groupId>
    	<artifactId>flink-table-planner-blink_2.12</artifactId>
    	<version>1.11.0</version>
    	<scope>provided</scope>
    </dependency>

Table API和SQL通过`join API`集成在一起，这个join API的核心概念是Table，Table可以作为查询的输入和输出。

##### Table API 和SQL的使用 #####

想要使用Table API 和SQL，首先要创建一个TableEnvironment对象。  
**Scala版本**

import org.apache.flink.api.scala.ExecutionEnvironment
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.bridge.scala.{ BatchTableEnvironment, StreamTableEnvironment}
    import org.apache.flink.table.api.{ EnvironmentSettings, TableEnvironment}
    
    /** * 创建TableEnvironment对象 */
    object CreateTableEnvironmentScala { 
      	def main(args: Array[String]): Unit = { 
    
    	    /** * 注意：如果Table API 和 SQL不需要和DataStream或者DataSet互相转换 * 则针对stream和batch都可以使用TableEnvironment */
    	    //指定底层引擎为Blink，以及数据处理模式-stream
    	    val sSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    	    //创建TableEnvironment对象
    	    val sTableEnv = TableEnvironment.create(sSettings)
    	
    	
    	    //指定底层引擎为Blink，以及数据处理模式-batch
    	    val bSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
    	    //创建TableEnvironment对象
    	    val bTableEnv = TableEnvironment.create(bSettings)
    	
    	    /** * 注意：如果Table API和SQL需要和DataStream或者DataSet互相转换 * 针对stream需要使用StreamTableEnvironment * 针对batch需要使用BatchTableEnvironment */
    	    //创建StreamTableEnvironment
    	    val ssEnv = StreamExecutionEnvironment.getExecutionEnvironment
    	    val ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    	    val ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings)
    	
    	    //创建BatchTableEnvironment
    	    //注意：此时只能使用旧的执行引擎，新的Blink执行引擎不支持和DataSet转换
    	    val bbEnv = ExecutionEnvironment.getExecutionEnvironment
    	    val bbTableEnv = BatchTableEnvironment.create(bbEnv)
    
      	}
    }

**Java版本**

import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.TableEnvironment;
    import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    /** * 创建TableEnvironment对象 */
    public class CreateTableEnvironmentJava { 
        public static void main(String[] args) { 
            /** * 注意：如果Table API 和 SQL不需要和DataStream或者DataSet互相转换 * 则针对stream和batch都可以使用TableEnvironment */
            //指定底层引擎为Blink，以及数据处理模式-stream
            EnvironmentSettings sSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            //创建TableEnvironment对象
            TableEnvironment sTableEnv = TableEnvironment.create(sSettings);
    
            //指定底层引擎为Blink，以及数据处理模式-batch
            EnvironmentSettings bSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
            //创建TableEnvironment对象
            TableEnvironment bTableEnv = TableEnvironment.create(bSettings);
    
            /** * 注意：如果Table API和SQL需要和DataStream或者DataSet互相转换 * 针对stream需要使用StreamTableEnvironment * 针对batch需要使用BatchTableEnvironment */
            //创建StreamTableEnvironment
            StreamExecutionEnvironment ssEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamTableEnvironment ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings);
    
            //创建BatchTableEnvironment
            ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnvironment();
            BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv);
        }
    }

下面来演示一下Table API和 SQL的使用  
目前创建Table的很多方法都过时了，都不推荐使用了，例如:`registerTableSource`、`connect`等方法  
目前官方推荐使用`executeSql`的方式 ， executeSql支持DDL、DML、DQL、SHOW、DESCRIBE、EXPLAIN、USE等语法

**Scala版本**

import org.apache.flink.table.api.{ EnvironmentSettings, TableEnvironment}
    
    /** * TableAPI 和 SQL的使用 */
    object TableAPIAndSQLOpScala { 
      	def main(args: Array[String]): Unit = { 
    	    //获取TableEnvironment
    	    val sSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    	    val sTableEnv = TableEnvironment.create(sSettings)
    	
    	    //创建输入表
    	    /** * connector.type：指定connector的类型 * connector.path：指定文件或者目录地址 * format.type：文件数据格式化类型，现在只支持csv格式 * 注意：SQL语句中如果出现了换行，行尾末尾可以添加空格或者\n都可以，最后一行不用添加 */
    	    sTableEnv.executeSql("" +
    	      "create table myTable(\n" +
    	      "id int,\n" +
    	      "name string\n" +
    	      ") with (\n" +
    	      "'connector.type' = 'filesystem',\n" +
    	      "'connector.path' = '~/data/source',\n" +
    	      "'format.type' = 'csv'\n" +
    	      ")")
    	
    	    //使用TableAPI实现数据查询和过滤等操作
    	    import org.apache.flink.table.api._
    	    /*val result = sTableEnv.from("myTable") .select($"id", $"name") .filter($"id" > 1)*/
    	
    	    //使用SQL实现数据查询和过滤等操作
    	    val result = sTableEnv.sqlQuery("select id,name from myTable where id > 1")
    	
    	    //输出结果到控制台
    	    result.execute().print()
    	
    	    //创建输出表
    	    sTableEnv.executeSql("" +
    	      "create table newTable(\n" +
    	      "id int,\n" +
    	      "name string\n" +
    	      ") with (\n" +
    	      "'connector.type' = 'filesystem',\n" +
    	      "'connector.path' = '~/data/res',\n" +
    	      "'format.type' = 'csv'\n" +
    	      ")")
    	
    	    //输出结果到表newTable中
    	    result.executeInsert("newTable")
      	}
    }

**Java版本**

import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.TableEnvironment;
    
    /** * TableAPI 和 SQL的使用 */
    public class TableAPIAndSQLOpJava { 
        public static void main(String[] args) { 
            //获取TableEnvironment对象
            EnvironmentSettings sSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            TableEnvironment sTableEnv = TableEnvironment.create(sSettings);
    
            //创建输入表
            sTableEnv.executeSql("" +
                    "create table myTable(\n" +
                    "id int,\n" +
                    "name string\n" +
                    ") with (\n" +
                    "'connector.type' = 'filesystem',\n" +
                    "'connector.path' = '~/data/source',\n" +
                    "'format.type' = 'csv'\n" +
                    ")");
    
            //使用TableAPI实现数据查询和过滤等操作
            /*Table result = sTableEnv.from("myTable") .select($("id"), $("name")) .filter($("id").isGreater(1));*/
    
            //使用SQL实现数据查询和过滤等操作
            Table result = sTableEnv.sqlQuery("select id,name from myTable where id > 1");
    
            //输出结果到控制台
            result.execute().print();
    
            //创建输出表
            sTableEnv.executeSql("" +
                    "create table newTable(\n" +
                    "id int,\n" +
                    "name string\n" +
                    ") with (\n" +
                    "'connector.type' = 'filesystem',\n" +
                    "'connector.path' = '~/data/res',\n" +
                    "'format.type' = 'csv'\n" +
                    ")");
    
            //输出结果到表newTable中
            result.executeInsert("newTable");
    
        }
    }

##### DataStream、DataSet和Table之间的互相转换 #####

Table API和SQL可以很容易的和DataStream和DataSet程序集成到一块。通过TableEnvironment ，可以把DataStream或者DataSet注册为Table，这样就可以使用Table API和SQL查询了。通过TableEnvironment 也可以把Table对象转换为DataStream或者DataSet，这样就可以使用DataStream或者DataSet中的相关API了。

1.  使用DataStream创建表，主要包含下面这两种情况

*  使用DataStream创建view视图
 *  使用DataStream创建table对象

**Scala版本**

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.EnvironmentSettings
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    
    /** * 将DataStream转换为表 */
    object DataStreamToTableScala { 
      	def main(args: Array[String]): Unit = { 
    	    //获取StreamTableEnvironment
    	    val ssEnv = StreamExecutionEnvironment.getExecutionEnvironment
    	    val ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    	    val ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings)
    	
    	    //获取DataStream
    	    import org.apache.flink.api.scala._
    	    val stream = ssEnv.fromCollection(Array((1, "jack"), (2, "tom"), (3, "mack")))
    	
    	    //第一种：将DataStream转换为view视图
    	    import org.apache.flink.table.api._
    	    ssTableEnv.createTemporaryView("myTable",stream,'id,'name)
    	    ssTableEnv.sqlQuery("select * from myTable where id > 1").execute().print()
    	
    	
    	    //第二种：将DataStream转换为Table对象
    	    val table = ssTableEnv.fromDataStream(stream, $"id", $"name")
    	    table.select($"id",$"name")
    	      .filter($"id" > 1)
    	      .execute()
    	      .print()
    	
    	    //注意：'id,'name 和 $"id", $"name" 这两种写法是一样的效果
      	}
    }

**Java版本**

import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    import java.util.ArrayList;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    /** * 将DataStream转换为表 */
    public class DataStreamToTableJava { 
        public static void main(String[] args) { 
            //获取StreamTableEnvironment
            StreamExecutionEnvironment ssEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamTableEnvironment ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings);
    
            //获取DataStream
            ArrayList<Tuple2<Integer, String>> data = new ArrayList<>();
            data.add(new Tuple2<Integer, String>(1,"jack"));
            data.add(new Tuple2<Integer, String>(2,"tom"));
            data.add(new Tuple2<Integer, String>(3,"mick"));
            DataStreamSource<Tuple2<Integer, String>> stream = ssEnv.fromCollection(data);
    
            //第一种：将DataStream转换为view视图
            ssTableEnv.createTemporaryView("myTable",stream,$("id"),$("name"));
            ssTableEnv.sqlQuery("select * from myTable where id > 1 ").execute().print();
    
            //第二种：将DataStream转换为table对象
            Table table = ssTableEnv.fromDataStream(stream, $("id"), $("name"));
            table.select($("id"),$("name"))
                    .filter($("id").isGreater(1))
                    .execute()
                    .print();
        }
    }

1.  使用DataSet创建表

> 注意:此时只能使用旧的执行引擎，新的Blink执行引擎不支持和DataSet转换

**Scala版本**

import org.apache.flink.api.scala.ExecutionEnvironment
    import org.apache.flink.table.api.bridge.scala.BatchTableEnvironment
    
    /** * 将DataSet转换为表 */
    object DataSetToTableScala { 
      	def main(args: Array[String]): Unit = { 
    	    //获取BatchTableEnvironment
    	    val bbEnv = ExecutionEnvironment.getExecutionEnvironment
    	    val bbTableEnv = BatchTableEnvironment.create(bbEnv)
    	
    	    //获取DataSet
    	    import org.apache.flink.api.scala._
    	    val set = bbEnv.fromCollection(Array((1, "jack"), (2, "tom"), (3, "mack")))
    	
    	    //第一种：将DataSet转换为view视图
    	    import org.apache.flink.table.api._
    	    bbTableEnv.createTemporaryView("myTable",set,'id,'name)
    	    bbTableEnv.sqlQuery("select * from myTable where id > 1").execute().print()
    	
    	    //第二种：将DataSet转换为table对象
    	    val table = bbTableEnv.fromDataSet(set, 'id, 'name)
    	    table.select($"id",$"name")
    	      .filter($"id" > 1 )
    	      .execute()
    	      .print()
      	}
    }

**Java版本**

import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.operators.DataSource;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
    
    import java.util.ArrayList;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    /** * 将DataSet转换为表 */
    public class DataSetToTableJava { 
        public static void main(String[] args) { 
            //获取BatchTableEnvironment
            ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnvironment();
            BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv);
    
            //获取DataSet
            //获取DataStream
            ArrayList<Tuple2<Integer, String>> data = new ArrayList<>();
            data.add(new Tuple2<Integer, String>(1,"jack"));
            data.add(new Tuple2<Integer, String>(2,"tom"));
            data.add(new Tuple2<Integer, String>(3,"mick"));
            DataSource<Tuple2<Integer, String>> set = bbEnv.fromCollection(data);
    
            //第一种：将DataSet转换为view视图
            bbTableEnv.createTemporaryView("myTable",set,$("id"),$("name"));
            bbTableEnv.sqlQuery("select * from myTable where id > 1 ").execute().print();
    
            //第二种：将DataSet转换为table对象
            bbTableEnv.fromDataSet(set,$("id"),$("name"))
                    .filter($("id").isGreater(1))
                    .execute()
                    .print();
        }
    }

将 Table 转换为 DataStream 或者 DataSet 时，你需要指定生成的 DataStream 或者 DataSet 的数据类型，即：Table 的每行数据要转换成的数据类型。通常最方便的选择是转换成 Row 。  
以下列表概述了不同选项的功能:

*  Row: 通过角标映射字段，支持任意数量的字段，支持 null 值，无类型安全(type-safe)检查。
 *  POJO: Java中的实体类，这个实体类中的字段名称需要和Table中的字段名称保持一致，支持任意数量的 字段，支持null值，有类型安全检查。
 *  Case Class: 通过角标映射字段，不支持null值，有类型安全检查。
 *  Tuple: 通过角标映射字段，Scala中限制22个字段，Java中限制25个字段，不支持null值，有类型安全检 查。
 *  Atomic Type: Table 必须有一个字段，不支持 null 值，有类型安全检查。

1.  将表转换成 DataStream

流式查询的结果Table会被动态地更新，即每个新的记录到达输入流时结果就会发生变化。因此，转换此动态查询的DataStream需要对表的更新进行编码。

有几种模式可以将Table转换为DataStream。

*  Append Mode:这种模式只适用于当动态表仅由INSERT更改修改时(仅附加)，之前添加的数据不会被更新。
 *  Retract Mode:可以始终使用此模式，它使用一个Boolean标识来编码INSERT和DELETE更改。

**Scala版本**

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.EnvironmentSettings
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    import org.apache.flink.types.Row
    
    /** * 将table转换成DataStream */
    object TableToDataStreamScala { 
      	def main(args: Array[String]): Unit = { 
    	    //获取StreamTableEnvironment
    	    val ssEnv = StreamExecutionEnvironment.getExecutionEnvironment
    	    val ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    	    val ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings)
    	
    	    //创建输入表
    	    ssTableEnv.executeSql("" +
    	      "create table myTable(\n" +
    	      "id int,\n" +
    	      "name string\n" +
    	      ") with (\n" +
    	      "'connector.type' = 'filesystem',\n" +
    	      "'connector.path' = '~/data/source',\n" +
    	      "'format.type' = 'csv'\n" +
    	      ")")
    	
    	    //获取table
    	    val table = ssTableEnv.from("myTable")
    	
    	    //将table转换为DataStream
    	    //如果只有新增(追加)操作，可以使用toAppendStream
    	    import org.apache.flink.api.scala._
    	    val appStream = ssTableEnv.toAppendStream[Row](table)
    	    appStream.map(row=>(row.getField(0).toString.toInt,row.getField(1).toString))
    	      .print()
    	
    	
    	    //如果有增加操作，还有删除操作，则使用toRetractStream
    	    val retStream = ssTableEnv.toRetractStream[Row](table)
    	    retStream.map(tup=>{ 
    	      val flag = tup._1
    	      val row = tup._2
    	      val id = row.getField(0).toString.toInt
    	      val name = row.getField(1).toString
    	      (flag,id,name)
    	    }).print()
    	
    	    //注意：将table转换为DataStream之后，就需要调用StreamExecutionEnvironment中的execute方法了
    	    ssEnv.execute("TableToDataStreamScala")
      }
    }

**Java版本**

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    import org.apache.flink.types.Row;
    
    /** * 将table转换成DataStream */
    public class TableToDataStreamJava { 
        public static void main(String[] args) throws Exception{ 
            //获取StreamTableEnvironment
            StreamExecutionEnvironment ssEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamTableEnvironment ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings);
    
            //创建输入表
            ssTableEnv.executeSql("" +
                    "create table myTable(\n" +
                    "id int,\n" +
                    "name string\n" +
                    ") with (\n" +
                    "'connector.type' = 'filesystem',\n" +
                    "'connector.path' = '~/data/source',\n" +
                    "'format.type' = 'csv'\n" +
                    ")");
    
            //获取table
            Table table = ssTableEnv.from("myTable");
    
            //将table转换为DataStream
            //如果只有新增(追加)操作，可以使用toAppendStream
            DataStream<Row> appStream = ssTableEnv.toAppendStream(table, Row.class);
            appStream.map(new MapFunction<Row, Tuple2<Integer,String>>() { 
                @Override
                public Tuple2<Integer, String> map(Row row) throws Exception { 
                    int id = Integer.parseInt(row.getField(0).toString());
                    String name = row.getField(1).toString();
                    return new Tuple2<Integer, String>(id,name);
                }
            }).print();
    
            //如果有增加操作，还有删除操作，则使用toRetractStream
            DataStream<Tuple2<Boolean, Row>> retStream = ssTableEnv.toRetractStream(table, Row.class);
            retStream.map(new MapFunction<Tuple2<Boolean, Row>, Tuple3<Boolean,Integer,String>>() { 
                @Override
                public Tuple3<Boolean, Integer, String> map(Tuple2<Boolean, Row> tup) throws Exception { 
                    Boolean flag = tup.f0;
                    int id = Integer.parseInt(tup.f1.getField(0).toString());
                    String name = tup.f1.getField(1).toString();
                    return new Tuple3<Boolean, Integer, String>(flag,id,name);
                }
            }).print();
    
            //注意：将table转换为DataStream之后，就需要调用StreamExecutionEnvironment中的execute方法了
            ssEnv.execute("TableToDataStreamJava");
        }
    }

1.  将表转换成 DataSet  
    **Scala版本**

import org.apache.flink.api.scala.ExecutionEnvironment
    import org.apache.flink.table.api.bridge.scala.BatchTableEnvironment
    import org.apache.flink.types.Row
    
    /** * 将table转换成DataSet */
    object TableToDataSetScala { 
      def main(args: Array[String]): Unit = { 
        //获取BatchTableEnvironment
        val bbEnv = ExecutionEnvironment.getExecutionEnvironment
        val bbTableEnv = BatchTableEnvironment.create(bbEnv)
    
        //创建输入表
        bbTableEnv.executeSql("" +
          "create table myTable(\n" +
          "id int,\n" +
          "name string\n" +
          ") with (\n" +
          "'connector.type' = 'filesystem',\n" +
          "'connector.path' = '~/data/source',\n" +
          "'format.type' = 'csv'\n" +
          ")")
    
        //获取table
        val table = bbTableEnv.from("myTable")
    
        //将table转换成DataSet
        import org.apache.flink.api.scala._
        val set = bbTableEnv.toDataSet[Row](table)
        set.map(row=>(row.getField(0).toString.toInt,row.getField(1).toString))
          .print()
    
      }
    }

**Java版本**

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
    import org.apache.flink.types.Row;
    
    /** * 将table转换成DataSet */
    public class TableToDataSetJava { 
        public static void main(String[] args) throws Exception{ 
            //创建BatchTableEnvironment
            ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnvironment();
            BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv);
    
            //创建输入表
            bbTableEnv.executeSql("" +
                    "create table myTable(\n" +
                    "id int,\n" +
                    "name string\n" +
                    ") with (\n" +
                    "'connector.type' = 'filesystem',\n" +
                    "'connector.path' = '~/data/source',\n" +
                    "'format.type' = 'csv'\n" +
                    ")");
    
            //获取table
            Table table = bbTableEnv.from("myTable");
    
            //将table转换成DataSet
            DataSet<Row> set = bbTableEnv.toDataSet(table, Row.class);
            set.map(new MapFunction<Row, Tuple2<Integer,String>>() { 
                @Override
                public Tuple2<Integer, String> map(Row row) throws Exception { 
                    int id = Integer.parseInt(row.getField(0).toString());
                    String name = row.getField(1).toString();
                    return new Tuple2<Integer, String>(id,name);
                }
            }).print();
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/d77ac578a6fc4028ad8013cbdffd6afa.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/07d454aad96e4e61b2d4d1b0f2639f7d.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/283d4b3e950d40728e3cd5d05a8adba4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/f9e2b5fd965d4a82b3127c1db2990760.png