Flink offers a distributed cache, similar to Apache Hadoop, to make files locally accessible to parallel instances of user functions. This functionality can be used to share files that contain static external data such as dictionaries or machine-learned regression models.           The cache works as follows. A program registers a file or directory of a local or remote filesystem such as HDFS or S3 under a specific name in its ExecutionEnvironment as a cached file. When the program is executed, Flink automatically copies the file or directory to the local filesystem of all workers. A user function can look up the file or directory under the specified name and access it from the worker’s local filesystem.

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// register a file from HDFSenv.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile")
// register a local executable file (script, executable, ...)env.registerCachedFile("file:///path/to/exec/file", "localExecFile", true)
// define your program and execute...DataStream input = ...DataStream result = input.map(new MyMapper());...env.execute();
---------------------------------------------------------------// extend a RichFunction to have access to the RuntimeContextpublic final class MyMapper extends RichMapFunction<String, Integer> {
    @Override    public void open(Configuration config) {
      // access cached file via RuntimeContext and DistributedCache      File myFile = getRuntimeContext().getDistributedCache().getFile("hdfsFile");      // read the file (or navigate the directory)      ...    }
    @Override    public Integer map(String value) throws Exception {      // use content of cached file      ...    }}

protected final ListString, DistributedCache.DistributedCacheEntry>> cacheFile = new ArrayList<>();
public void registerCachedFile(String filePath, String name, boolean executable) {    this.cacheFile.add(new Tuple2<>(name, new DistributedCache.DistributedCacheEntry(filePath, executable)));}

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment#getStreamGraphGenerator   private StreamGraphGenerator getStreamGraphGenerator() {    if (transformations.size() <= 0) {        throw new IllegalStateException("No operators defined in streaming topology. Cannot execute.");    }    return new StreamGraphGenerator(transformations, config, checkpointCfg)        .setStateBackend(defaultStateBackend)        .setChaining(isChainingEnabled)        .setUserArtifacts(cacheFile)   // note:傳遞cacheFile        .setTimeCharacteristic(timeCharacteristic)        .setDefaultBufferTimeout(bufferTimeout);}
org.apache.flink.streaming.api.graph.StreamGraphGenerator#generatepublic StreamGraph generate() {    streamGraph = new StreamGraph(executionConfig, checkpointConfig, savepointRestoreSettings);    streamGraph.setStateBackend(stateBackend);    streamGraph.setChaining(chaining);    streamGraph.setScheduleMode(scheduleMode);    streamGraph.setUserArtifacts(userArtifacts); // note:傳遞userArtifacts    streamGraph.setTimeCharacteristic(timeCharacteristic);    streamGraph.setJobName(jobName);    streamGraph.setBlockingConnectionsBetweenChains(blockingConnectionsBetweenChains);
    alreadyTransformed = new HashMap<>();
    for (Transformation transformation: transformations) {        transform(transformation);    }
    final StreamGraph builtStreamGraph = streamGraph;
    alreadyTransformed.clear();    alreadyTransformed = null;    streamGraph = null;
    return builtStreamGraph;}

org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator#createJobGraph()    private JobGraph createJobGraph() {    ...    JobGraphGenerator.addUserArtifactEntries(streamGraph.getUserArtifacts(), jobGraph);    ...    return jobGraph;}
public static void addUserArtifactEntries(Collection> userArtifacts, JobGraph jobGraph) {    if (userArtifacts != null && !userArtifacts.isEmpty()) {        try {            java.nio.file.Path tmpDir = Files.createTempDirectory("flink-distributed-cache-" + jobGraph.getJobID());            for (Tuple2 originalEntry : userArtifacts) {                Path filePath = new Path(originalEntry.f1.filePath);                boolean isLocalDir = false;                try {                    FileSystem sourceFs = filePath.getFileSystem();                    isLocalDir = !sourceFs.isDistributedFS() && sourceFs.getFileStatus(filePath).isDir();                } catch (IOException ioe) {                    LOG.warn("Could not determine whether {} denotes a local path.", filePath, ioe);                }                // zip local directories because we only support file uploads                DistributedCache.DistributedCacheEntry entry;                if (isLocalDir) {                    // note: 壓縮本地文件夾，返回zip文件路徑                    Path zip = FileUtils.compressDirectory(filePath, new Path(tmpDir.toString(), filePath.getName() + ".zip"));                    entry = new DistributedCache.DistributedCacheEntry(zip.toString(), originalEntry.f1.isExecutable, true);                } else {                    entry = new DistributedCache.DistributedCacheEntry(filePath.toString(), originalEntry.f1.isExecutable, false);                }                jobGraph.addUserArtifact(originalEntry.f0, entry);            }        } catch (IOException ioe) {            throw new FlinkRuntimeException("Could not compress distributed-cache artifacts.", ioe);        }    }}

org.apache.flink.yarn.YarnClusterDescriptor#startAppMaster
// only for per job modeif (jobGraph != null) {    for (Map.Entry entry : jobGraph.getUserArtifacts().entrySet()) {        org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(entry.getValue().filePath);        // only upload local files        // note: 上傳本地文件，返回hdfs中的路徑存儲在jobGraph的userArtifacts        if (!path.getFileSystem().isDistributedFS()) {            Path localPath = new Path(path.getPath());            Tuple2 remoteFileInfo =                Utils.uploadLocalFileToRemote(fs, appId.toString(), localPath, homeDir, entry.getKey());            jobGraph.setUserArtifactRemotePath(entry.getKey(), remoteFileInfo.f0.toString());        }    }    // 將分布式緩存文件信息寫入到Configuration中    jobGraph.writeUserArtifactEntriesToConfiguration();}

DistributedCache#writeFileInfoToConfigpublic static void writeFileInfoToConfig(String name, DistributedCacheEntry e, Configuration conf) {    int num = conf.getInteger(CACHE_FILE_NUM, 0) + 1;    conf.setInteger(CACHE_FILE_NUM, num);    conf.setString(CACHE_FILE_NAME + num, name);    // note: DISTRIBUTED_CACHE_FILE_PATH_0    conf.setString(CACHE_FILE_PATH + num, e.filePath);    conf.setBoolean(CACHE_FILE_EXE + num, e.isExecutable || new File(e.filePath).canExecute());    conf.setBoolean(CACHE_FILE_DIR + num, e.isZipped || new File(e.filePath).isDirectory());    if (e.blobKey != null) {        conf.setBytes(CACHE_FILE_BLOB_KEY + num, e.blobKey);    }}

org.apache.flink.runtime.taskmanager.Task#doRunprivate void doRun() {    .......    // all resource acquisitions and registrations from here on    // need to be undone in the end    Map> distributedCacheEntries = new HashMap<>();
    // next, kick off the background copying of files for the distributed cache    try {        for (Map.Entry entry :                DistributedCache.readFileInfoFromConfig(jobConfiguration)) {            LOG.info("Obtaining local cache file for '{}'.", entry.getKey());                        Future cp = fileCache.createTmpFile(entry.getKey(), entry.getValue(), jobId, executionId);            distributedCacheEntries.put(entry.getKey(), cp);        }    }    catch (Exception e) {        throw new Exception(            String.format("Exception while adding files to distributed cache of task %s (%s).", taskNameWithSubtask, executionId), e);    }
    Environment env = new RuntimeEnvironment(        jobId,        vertexId,        executionId,        executionConfig,        taskInfo,        jobConfiguration,        taskConfiguration,        userCodeClassLoader,        memoryManager,        ioManager,        broadcastVariableManager,        taskStateManager,        aggregateManager,        accumulatorRegistry,        kvStateRegistry,        inputSplitProvider,        distributedCacheEntries,  // note:         consumableNotifyingPartitionWriters,        inputGates,        taskEventDispatcher,        checkpointResponder,        taskManagerConfig,        metrics,        this);}

public Future createTmpFile(String name, DistributedCacheEntry entry, JobID jobID, ExecutionAttemptID executionId) throws Exception {    synchronized (lock) {        Map> jobEntries = entries.computeIfAbsent(jobID, k -> new HashMap<>());
        // register reference holder        final Set refHolders = jobRefHolders.computeIfAbsent(jobID, id -> new HashSet<>());        refHolders.add(executionId);
        Future fileEntry = jobEntries.get(name);        if (fileEntry != null) {            // file is already in the cache. return a future that            // immediately returns the file            return fileEntry;        } else {            // need to copy the file                        // create the target path            File tempDirToUse = new File(storageDirectories[nextDirectory++], jobID.toString());            if (nextDirectory >= storageDirectories.length) {                nextDirectory = 0;            }
            // kick off the copying            Callable cp;            if (entry.blobKey != null) {                cp = new CopyFromBlobProcess(entry, jobID, blobService, new Path(tempDirToUse.getAbsolutePath()));            } else {                ## note: 從hdfs異步拷貝到TM內(nèi)部文件夾                cp = new CopyFromDFSProcess(entry, new Path(tempDirToUse.getAbsolutePath()));            }            FutureTask copyTask = new FutureTask<>(cp);            executorService.submit(copyTask);
            // store our entry            jobEntries.put(name, copyTask);
            return copyTask;        }    }}

org.apache.flink.api.common.cache.DistributedCache#getFile
public File getFile(String name) {    // note: Map> distributedCacheEntries     Future future = cacheCopyTasks.get(name);
    try {        final Path path = future.get();        URI tmp = path.makeQualified(path.getFileSystem()).toUri();        return new File(tmp);    }    catch (ExecutionException e) {        throw new RuntimeException("An error occurred while copying the file.", e.getCause());    }    catch (Exception e) {        throw new RuntimeException("Error while getting the file registered under '" + name +                "' from the distributed cache", e);    }}