Flink中-C参数与DistributedCache使用

-C 参数

flink命令下有这样的一个参数选项-C或者--classpath，含义解释：

Adds a URL to each user code classloader  on all nodes in 
the cluster. The paths must specify a protocol (e.g. file://)
 and be accessible on all nodes (e.g. by means of a NFS 
 share). You can use this option multiple times for 
 specifying more than one URL. The protocol must be 
 supported by the {@link java.net.URLClassLoader}.

含义就是：给一个user classloader添加一个url，但是这个url 必须能够被集群的所有的节点都能够访问到。该classloader指的是FlinkUserCodeClassLoaders，在任务启动的过程中会使用该loader加载，具体使用是在StreamTask.invoke中初始化OperatorChain中，在OperatorChain初始化时，会从字节码中反序列化一个operator的header operator，在这个加载过程中会使用FlinkUserCodeClassLoaders进行加载，

final ClassLoader userCodeClassloader = containingTask.getUserCodeClassLoader();
final StreamConfig configuration = containingTask.getConfiguration();
headOperator = configuration.getStreamOperator(userCodeClassloader);

FlinkUserCodeClassLoaders分ParentFirstClassLoader与ChildFirstClassLoader两类，默认使用ChildFirstClassLoader，表示在加载过程中会优先从给定的url中加载类。

StreamOperator面向用户调用的就是UserFunction，如果我们的自定义Function中有一些比较通用的包，有很多Flink任务都会使用到，那么我们就可以使用-C 来指定包的路径，前提是集群的每个node都可以访问到(file://)，通过这种方式程序在打包的时候就不需要将这些通用的包打进去。

DistributedCache

DistributedCache正如其含义分布式缓存，其功能与spark的广播变量类似，仅仅只会在一个TaskExecutor中维护一份该数据，用法：

 //注册
 env.registerCachedFile("file:/1.log","file1")

在userFunction中：

val file=getRuntimeContext.getDistributedCache.getFile("file1")

在ha模式下的工作机制：文件会被上传到high-availability.storageDir指定的目录下(一般是hdfs)，在任务启动过程中，会启动一个后台线程从hdfs拉取文件到本地可提供访问。

在非ha默认下的工作机制：文件存储在jobmaster节点下的工作路径中，在任务启动过程中从jobmaster中拉取文件到taskExecutor本地路径下。

关注回复 Flink

获取更多系列