分布式缓存不起作用

Question

我将少量数据（几 MB）存储在分布式缓存中，并使用它对两个大文件执行反连接。对于缓存中的几行数据，该功能运行良好，但是当缓存在生产中有更多数据时，它无法完成这项工作，但也不会引发任何错误。只是只有少数记录（约 20%）被加入，而其他记录则被忽略。那么分布式缓存中可以存储的记录数有上限吗？为什么它为某些记录工作而忽略其余记录？任何建议都会非常有帮助。 下面是我的代码

        public class MyMapper extends Mapper<LongWritable, Text, Text, TextPair> {

            Text albumKey = new Text();
            Text photoKey = new Text();
            private HashSet<String> photoDeleted = new HashSet<String>();

            private HashSet<String> albDeleted = new HashSet<String>();
            Text interKey = new Text();
            private TextPair interValue = new TextPair();
            private static final Logger LOGGER = Logger.getLogger(SharedStreamsSlMapper.class);

            protected void setup(Context context) throws IOException, InterruptedException {
                int count=0;
                Path[] cacheFiles = DistributedCache.getLocalCacheFiles(context.getConfiguration());
                System.out.println(cacheFiles.length);
                LOGGER.info(cacheFiles+"****");
                try {
                    if (cacheFiles != null && cacheFiles.length > 0) {
                        for (Path path : cacheFiles) {
                            String line;
                            String[] tokens;

                            BufferedReader joinReader = new BufferedReader(new FileReader(path.toString()));
                            System.out.println(path.toString());
        //                  BufferedReader joinReader = new BufferedReader(new FileReader("/Users/Kunal_Basak/Desktop/ss_test/dsitCache/part-m-00000"));
                            try {
                                while ((line = joinReader.readLine()) != null) {
                                    count++;
                                    tokens = line.split(SSConstants.TAB, 2);
                                    if(tokens.length<2){
                                        System.out.println("WL");
                                        continue;
                                    }
                                    if (tokens[0].equals("P")) {
                                        photoDeleted.add(tokens[1]);
                                    }
                                    else if (tokens[0].equals("A")) {
                                        albDeleted.add(tokens[1]);
                                    }
                                }
                            }
                            finally {
                                joinReader.close();
                            }
                        }
                    }
                }
                catch (IOException e) {
                    System.out.println("Exception reading DistributedCache: " + e);
                }
                System.out.println(count);
                System.out.println("albdeleted *****"+albDeleted.size());
                System.out.println("photo deleted *****"+photoDeleted.size());
                LOGGER.info("albdeleted *****"+albDeleted.size());
                LOGGER.info("albdeleted *****"+albDeleted.size());
            }

            public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        try{
    //my mapper code
    }
    }
    }

Answer 1

据此blog article：

local.cache.size 参数控制 分布式缓存。

默认设置为 10 GB。

因此，如果您的缓存超过 10GB，那可能是您的问题。