Flink扩展API

小咪咪 2023-02-14 01:19 53阅读 0赞

## 一.简介 ##

为了在Scala和Java API之间保持相当程度的一致性，用于批处理和流传输的标准API省略了一些允许在Scala中进行高水平表达的功能。  
如果想享受完整的Scala体验，则可以选择加入通过隐式转换来增强Scala API的扩展。  
要使用所有可用的扩展，只需导入相应的扩展组件即可：  
1.DataSet API

import org.apache.flink.api.scala.extensions._

2.DataStream API

import org.apache.flink.streaming.api.scala.extensions._

另外，也可以根据需要导入单个扩展。

## 二.扩展模式匹配 ##

通常，DataSet和DataStream API都不接受匿名模式匹配函数来匹配元组，案例类或集合，如下所示：

val data: DataSet[(Int, String, Double)] = // [...]
    data.map { 
      case (id, name, temperature) => // [...]
      // The previous line causes the following compilation error:
      // "The argument types of an anonymous function must be fully known. (SLS 8.5)"
    }

此扩展在DataSet和DataStream Scala API中引入了新方法，这些新方法在扩展API中具有一对一的对应关系。这些扩展方法支持匿名模式匹配功能。  
1.DataSet API  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]  
DataSet要专门使用此扩展，可以添加以下内容：

import org.apache.flink.api.scala.extensions.acceptPartialFunctions

2.DataStream API  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][20200530090306838.png]  
DataStream要专门使用此扩展，可以添加以下内容：

import org.apache.flink.streaming.api.scala.extensions.acceptPartialFunctions

## 三.代码案例 ##

以下代码片段显示了如何一起使用这些扩展方法（与DataSet API一起使用）的示例：

package cn.extensions
    
    import org.apache.flink.api.scala._
    import org.apache.flink.api.scala.ExecutionEnvironment
    
    /** * Created by Administrator on 2020/5/29. */
    case class Person(x: String, y: Int)
    object Match { 
      def main(args: Array[String]): Unit = { 
        // 设置execution执行环境
        val env = ExecutionEnvironment.getExecutionEnvironment
    
        val text = "Apache Flink apache spark apache solr hbase hive flink kafka redis tachyon redis"
        val persons = text.toLowerCase.split(" ").map(row => Person(row, 1))
        
        import org.apache.flink.api.scala.extensions._
        val ds = env.fromCollection(persons)
        val result = ds.filterWith { 
          case Person(x, y) => y > 0
        }.groupingBy{ 
          case Person(x, _) => x
        }.sum("y")
    
        result.first(5).print()
      }
    }

异常报错信息：

Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.
    	at org.apache.flink.api.scala.operators.ScalaAggregateOperator.translateToDataFlow(ScalaAggregateOperator.java:220)
    	at org.apache.flink.api.scala.operators.ScalaAggregateOperator.translateToDataFlow(ScalaAggregateOperator.java:55)
    	at org.apache.flink.api.java.operators.OperatorTranslation.translateSingleInputOperator(OperatorTranslation.java:148)
    	at org.apache.flink.api.java.operators.OperatorTranslation.translate(OperatorTranslation.java:102)
    	at org.apache.flink.api.java.operators.OperatorTranslation.translateSingleInputOperator(OperatorTranslation.java:146)
    	at org.apache.flink.api.java.operators.OperatorTranslation.translate(OperatorTranslation.java:102)
    	at org.apache.flink.api.java.operators.OperatorTranslation.translate(OperatorTranslation.java:63)
    	at org.apache.flink.api.java.operators.OperatorTranslation.translateToPlan(OperatorTranslation.java:52)
    	at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:955)
    	at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:922)
    	at org.apache.flink.api.java.LocalEnvironment.execute(LocalEnvironment.java:85)
    	at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:816)
    	at org.apache.flink.api.java.DataSet.collect(DataSet.java:413)
    	at org.apache.flink.api.java.DataSet.print(DataSet.java:1652)
    	at org.apache.flink.api.scala.DataSet.print(DataSet.scala:1726)
    	at cn.extensions.Match$.main(Match.scala:29)
    	at cn.extensions.Match.main(Match.scala)
    	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    	at java.lang.reflect.Method.invoke(Method.java:497)
    	at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)

可知是使用groupingWith导致的，意思是不支持使用该扩展API，换回原来的groupWith即可：  
![在这里插入图片描述][20200530163500967.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200530085909215.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/202005300859496.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200530090009828.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200530090221324.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200530090246585.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[20200530090306838.png]: https://img-blog.csdnimg.cn/20200530090306838.png
[20200530163500967.png]: https://img-blog.csdnimg.cn/20200530163500967.png