1. devBase\WithParameters.scala

package devBase 
 
import org.apache.flink.api.common.functions.RichMapFunction 
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, createTypeInformation} 
import org.apache.flink.configuration.{ConfigOptions, Configuration} 
 
 
 
class WithParametersMapFunction extends RichMapFunction[String, String] { 
 
  private var bigdata:String = null 
 
  override def map(value: String):String = { 
 
    if (value == bigdata) "bigdata" else value 
  } 
 
  override def open(parameters: Configuration): Unit = { 
 
    val configOption = ConfigOptions.key("bigdata") 
      .stringType() 
      .noDefaultValue() 
 
    bigdata = parameters.getString(configOption) 
 
  } 
 
} 
 
object WithParameters { 
 
 
  def main(args: Array[String]): Unit = { 
 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
    val conf = new Configuration() 
    conf.setString("bigdata", "flink") 
 
    val text: DataSet[String] = env.fromElements("flink", "spark", "hadoop") 
 
    val ds = text.map(new WithParametersMapFunction()) 
      // 只支持batch处理; 作用于上面的map, 将conf传递给rich类的open方法 
      .withParameters(conf) 
 
    ds.print() 
 
 
  } 
 
} 

执行结果如下：

bigdata 
spark 
hadoop 

2. devBase\GlobalJobParameters.scala

package devBase 
 
import apiTest.WordSourceFunction 
import org.apache.flink.api.common.functions.RichMapFunction 
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, createTypeInformation} 
import org.apache.flink.configuration.{ConfigOptions, Configuration} 
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} 
 
 
class GlobalJobParametersMapFunction extends RichMapFunction[String, String] { 
 
  private var bigdata:String = null 
 
  override def map(value: String):String = { 
 
    if (value == bigdata) "bigdata-" + value else value 
  } 
 
  override def open(parameters: Configuration): Unit = { 
 
    val globalJobParameters= getRuntimeContext 
      .getExecutionConfig.getGlobalJobParameters 
      .asInstanceOf[Configuration] 
 
    val configOption = ConfigOptions.key("bigdata") 
      .stringType() 
      .noDefaultValue() 
 
    bigdata = globalJobParameters.getString(configOption) 
 
  } 
 
} 
 
object GlobalJobParameters { 
 
  def main(args: Array[String]): Unit = { 
 
    val conf = new Configuration() 
    conf.setString("bigdata", "flink") 
    val senv = StreamExecutionEnvironment.getExecutionEnvironment 
    senv.getConfig.setGlobalJobParameters(conf) 
    // senv.getConfig.setGlobalJobParameters(parameterTool) 
 
 
    val text: DataStream[String] = senv.addSource(new WordSourceFunction) 
 
    val ds = text.map(new GlobalJobParametersMapFunction()) 
 
 
    ds.print() 
    senv.execute() 
 
 
 
  } 
 
} 

执行结果：

4> world 
5> stream 
6> table 
7> sql 
8> bigdata-flink 
1> bigdata-flink 
2> batch 
......省略部分...... 

3. devBase\ParameterToolTest.scala

package devBase 
 
import org.apache.flink.api.java.utils.ParameterTool 
 
import java.io.{File, FileInputStream} 
import scala.collection.JavaConversions.mapAsJavaMap 
 
 
object ParameterToolTest { 
 
 
  def main(args: Array[String]): Unit = { 
 
    val map = Map("id" -> "10", "name" -> "LiMing", "age" -> "10") 
    val parameter_map = ParameterTool.fromMap(map) 
    println("======map====== " + parameter_map.getRequired("name")) 
    println("======map====== " + parameter_map.get("name", "default_name")) 
    println("======map====== " + parameter_map.getInt("id", 0)) 
    println("======map====== " + parameter_map.getNumberOfParameters) 
    println("======map====== " + parameter_map.getProperties) 
 
 
    val prop_path = "src/main/resources/parameterToolTest.properties" 
    val parameter_prop_path=ParameterTool.fromPropertiesFile(prop_path) 
    println("======path====== " + parameter_prop_path.get("name")) 
 
    val prop_file = new File(prop_path) 
    val parameter_prop_file=ParameterTool.fromPropertiesFile(prop_file) 
    println("======file====== " + parameter_prop_file.get("name")) 
 
    val prop_stream = new FileInputStream(prop_file) 
    val parameter_prop_stream=ParameterTool.fromPropertiesFile(prop_stream) 
    println("======stream====== " + parameter_prop_stream.get("name")) 
 
    // 例子：--input hdfs://xxx --output hdfs://xxx 
    val parameter_args=ParameterTool.fromArgs(args) 
    println("======args====== " + parameter_args.get("name")) 
 
    val parameter_system=ParameterTool.fromSystemProperties() 
    println("======system====== " + parameter_system.get("java.vm.version")) 
 
 
  } 
 
} 

执行结果：

======map====== LiMing 
======map====== LiMing 
======map====== 10 
======map====== 3 
======map====== {age=10, name=LiMing, id=10} 
======path====== LiMing 
======file====== LiMing 
======stream====== LiMing 
======args====== null 
======system====== 25.201-b09 

4. ParameterTool解决中文乱码问题

使用ParameterTool读取properties配置文件，如果有中文，读取的是会乱码的。可以使用如下方法指定UTF-8编码读取

import java.io.{BufferedReader, InputStream, InputStreamReader} 
import java.util.Properties 
import org.apache.flink.api.java.utils.ParameterTool 
 
object ParameterToolTest { 
 
  def main(args: Array[String]): Unit = { 
   
    val resourceProperties = new Properties() 
    val resourceInputStream:InputStream = this.getClass.getClassLoader.getResourceAsStream("parameterToolTest.properties") 
    val resourceInputStreamReader:InputStreamReader = new InputStreamReader(resourceInputStream, "UTF-8") 
    val resourceBufferedReader:BufferedReader = new BufferedReader(resourceInputStreamReader) 
    resourceProperties.load(resourceBufferedReader) 
 
    val parameterTool = ParameterTool.fromMap(resourceProperties.asInstanceOf[java.util.Map[String,String]]) 
    println("======chinese stream====== " + parameterTool.get("chinese_name")) 
  } 
 
} 

执行结果：

======chinese stream====== 李明