Flink传递参数给函数

原文地址&＃xff1a;https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/batch/index.html#passing-parameters-to-functions

Passing Parameters to Functions

  1 使用构造函数方式:

package com.daxinimport org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
/*** Created by Daxin on 2017/4/18.*/
object PassingParameters2Functions1 {def main(args: Array[String]) {val env &＃61; ExecutionEnvironment.getExecutionEnvironmentval toFilter &＃61; env.fromElements(1, 2, 3)class MyFilter(limit: Int) extends FilterFunction[Int] {override def filter(value: Int): Boolean &＃61; {value > limit}}val result &＃61;toFilter.filter(new MyFilter(2))result.print()}}


 2 withParameters(Configuration)方式

这个方法将会携带一个Configuration对象作为参数&＃xff0c;这个参数将会传递给Rich Function的open方法(关于Rich Function参见&＃xff1a;rich function)。Configuration对象是一个Map&＃xff0c;存储Key/Value键值对.

package com.daxinimport org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration/*** Created by Daxin on 2017/4/18.* 我的邮箱: leodaxin&＃64;163.com*/
object PassingParameters2Functions2 {def main(args: Array[String]) {val env &＃61; ExecutionEnvironment.getExecutionEnvironmentval toFilter &＃61; env.fromElements(1, 2, 3)val c &＃61; new Configuration()c.setInteger("limit", 2)val result &＃61; toFilter.filter(new RichFilterFunction[Int]() {var limit &＃61; 0override def open(config: Configuration): Unit &＃61; {limit &＃61; config.getInteger("limit", 0) //没有的话返回默认值0}def filter(in: Int): Boolean &＃61; {in > limit}}).withParameters(c)result.print()}}

package com.daxinimport org.apache.flink.api.common.ExecutionConfig
import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction, RichFilterFunction}
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration/*** Created by Daxin on 2017/4/18.* 传递参数3&＃xff1a;Globally via the ExecutionConfig**/
object PassingParameters2Functions3 {def main(args: Array[String]) {val env &＃61; ExecutionEnvironment.getExecutionEnvironmentval data &＃61; env.fromElements("1", "2")val conf &＃61; new Configuration()conf.setString("mykey", "2")env.getConfig.setGlobalJobParameters(conf)class RichFunc extends RichMapFunction[String, String] {var mykey: String &＃61; _override def open(parameters: Configuration): Unit &＃61; {super.open(parameters)val globalParams: ExecutionConfig.GlobalJobParameters &＃61; getRuntimeContext().getExecutionConfig().getGlobalJobParameters()val globConf &＃61; globalParams.asInstanceOf[Configuration]mykey &＃61; globConf.getString("mykey", "default")}override def map(value: String): String &＃61; {if (mykey.equals(value)) "is equals" else "not equals"}}data.map(new RichFunc).print()}}