解析命令行参数和在Flink应用程序中传递参数
几乎所有的Flink应用程序，包括批处理和流处理，都依赖于外部配置参数，这些参数被用来指定输入和输出源(如路径或者地址)，系统参数(并发数，运行时配置)和应用程序的可配参数(通常用在自定义函数中)。
Flink提供了一个简单的叫做ParameterTool的utility，ParameterTool提供了一些基础的工具来解决这些问题，当然你也可以不用这里所有描述的ParameterTool，其他框架如:Commons CLI和argparse4j在Flink中也是支持的。
获取你的配置值并传入ParameterTool中
ParameterTool提供了一系列预定义的静态方法来读取配置信息，ParameterTool内部是一个Map<String, String>,所以很容易与你自己的配置形式相集成。
从.properties文件中获取
下面方法将去读取一个Properties文件，并返回若干key/value对:
String propertiesFile = "/home/sam/flink/myjob.properties";
ParameterTool parameter = ParameterTool.fromPropertiesFile(propertiesFile);
从命令行参数中获取
下面会从命令行中获取像--input hdfs:///mydata --elements 42这种形式的参数:
public static void main(String[] args) {
ParameterTool parameter = ParameterTool.fromArgs(args);
// .. regular code ..
从系统属性中获取
当启动一个JVM时，你可以给它传递一些系统属性如:-Dinput=hdfs:///mydata，你也可以用这些系统属性来初始化ParameterTool:
ParameterTool parameter = ParameterTool.fromSystemProperties();
在程序中使用ParameterTool的参数
既然我们已经从其他地方(方法如上)拿到了配置参数，我们就可以以各种形式来使用它们了。
直接从ParameterTool中获取
ParameterTool本身有方法来获取这些值:
ParameterTool parameters = // ...
parameter.getRequired("input");
parameter.get("output", "myDefaultValue");
parameter.getLong("expectedCount", -1L);
parameter.getNumberOfParameters()
// .. there are more methods available.
你可以在提交应用程序的客户端main()方法中直接使用这些方法返回的值，例如:你可以按如下方法来设置一个算子的并发度:
ParameterTool parameters = ParameterTool.fromArgs(args);
int parallelism = parameters.get("mapParallelism", 2);
DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).setParallelism(parallelism);
因为ParameterTool是可序列化的，所以你可以将它传递给函数本身；
ParameterTool parameters = ParameterTool.fromArgs(args);
DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer(parameters));
之后在函数内部使用ParameterTool来获取命令行参数。
将参数以Configuration对象的形式传递给函数
下面的例子展示了如何将参数以Configuration对象的形式传递给用户自定义函数。
ParameterTool parameters = ParameterTool.fromArgs(args);
DataSet<Tuple2<String, Integer>> counts = text
.flatMap(new Tokenizer()).withParameters(parameters.getConfiguration())
在Tokenizer内部，Configuration对象可以通过open(Configuration conf)方法来获取；
public static final class Tokenizer extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void open(Configuration parameters) throws Exception {
parameters.getInteger("myInt", -1);
// .. do
注册全局参数
在ExecutionConfig中注册为全作业参数的参数，可以被JobManager的web端以及用户自定义函数中以配置值的形式访问.
注册全局参数:
ParameterTool parameters = ParameterTool.fromArgs(args);
// 创建一个执行环境
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(parameters);
在用户自定义的富函数中获取它们:
public static final class Tokenizer extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
ParameterTool parameters = (ParameterTool)
getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
parameters.getRequired("input");
// .. do more ..
命名大型的TupleX类型
对于有很多字段的数据类型我们建议采用POJOs(普通Java对象)，而不是TupleX；同时，POJOs还可以用来为大型的Tuple命名。、
例如：
不使用:Tuple11<String, String, ..., String> var = new ...;
因为继承大型Tuple类来创建一个自定义的类型会比直接使用大型Tuple简单得多:
CustomType var = new ...;
public static class CustomType extends Tuple11<String, String, ..., String> {
// constructor matching super
}
使用Logback而不是Log4j
注意:本文档适用于Flink 0.10之后的版本
Apache Flink使用slf4j来作为logging 抽象，也建议用户在他们自定义的方法中也使用slf4j。Slf4j是一个编译时的logging接口，在运行时可以使用不同的logging实现，例如:log4j或者Logback。
默认情况下Flink依赖Log4j，本也描述了如何在Flink中使用Logback。有用户反馈它们通过本指南，可以使用Graylog来建立集中式日志收集。
使用下面的代码来获取一个logger实例:
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MyClass implements MapFunction {
private static final Logger LOG = LoggerFactory.getLogger(MyClass.class);
// ...
在IDE之外运行Flink时使用Logback或者在一个Java应用程序中使用Logback
在所有情况下类会在一个由依赖管理器如Maven创建的classpath中执行，Flink会将log4j推到classpath中。
因此，你需要将log4j从Flink的依赖中剔除，下面的描述假定有一个跟Maven工程:
按如下方式来修改你的工程:
<dependencies>
<!-- Add the two required logback dependencies -->
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
<version>1.1.3</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.3</version>
</dependency>
<!-- Add the log4j -> sfl4j (-> logback) bridge into the classpath
Hadoop is logging to log4j! -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>log4j-over-slf4j</artifactId>
<version>1.7.7</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.3.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.10</artifactId>
<version>1.3.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.10</artifactId>
<version>1.3.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
在<dependencies>部分进行如下修改；
1、将所有的log4j依赖从所有的Flink dependencies中剔除:这会导致Maven忽略Flink对log4j的传递依赖。
2、将slf4j-log4j12 artifact从Flink依赖中剔除:因为我们将用slf4j到logback的绑定，所以我们需要删除slf4j到log4j的绑定。
3、添加Logback依赖:logback-core和logback-classic
4、添加log4j-over-slf4j依赖:log4j-over-slf4j是一种允许旧应用程序直接使用Log4j API来调用Slf4j接口的工具。Flink依赖Hadoop，而Hadoop是使用Log4j来记录日志的
请注意:你需要手动添加exclusion到所有你添加到pom文件中的Flink依赖。
你可能还需要检查一下其他非Flink依赖是否也是log4j的绑定，你可以使用mvn dependency:true来分析你的工程依赖。
当Flink运行在集群中是使用Logback
本指南适用于当Flink运行在YARN或者standalong集群时。
为了在Flink中使用Logback而不是Log4j，你需要将 log4j-1.2.xx.jar 和 sfl4j-log4j12-xxx.jar从 lib/目录中删除
接下来，你需要将下面的jar文件添加到 lib/目录下:
logback-classic.jar
logback-core.jar
log4j-over-slf4j.jar
注意:当使用单任务的YARN集群时，你需要明确的设置 lib/目录!
将自定义logger的Flink提交到YARN中的命令是:./bin/flink run -yt $FLINK_HOME/lib <... remaining arguments ...>
