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本地路径下。

—END—

关注回复 Flink

获取更多系列

原创不易,好看,就点个"在看"

我来评几句
登录后评论

已发表评论数()

相关站点

热门文章