@Override
  public HeuristicResult apply(MapReduceApplicationData data) {
    if (data == null || !data.getSucceeded()) {
      return null;
    }

    Properties jobConf = data.getConf();
    String cacheFiles = jobConf.getProperty(MAPREDUCE_JOB_CACHE_FILES, null);
    String cacheFileSizes = jobConf.getProperty(MAPREDUCE_JOB_CACHE_FILES_FILESIZES, null);

    HeuristicResult result = null;

    if (cacheFiles != null && cacheFileSizes != null) {
      result =
          new HeuristicResult(_heuristicConfData.getClassName(), _heuristicConfData.getHeuristicName(), Severity.NONE,
              0);
      List<String> cacheFilesList = new ArrayList<String>(Arrays.asList(cacheFiles.split(",")));
      List<String> cacheFileSizesList = new ArrayList<String>(Arrays.asList(cacheFileSizes.split(",")));

      int cacheFilesCount = cacheFilesList.size();
      int cacheFileSizesCount = cacheFileSizesList.size();

      if (cacheFilesCount != cacheFileSizesCount) {
        result.setSeverity(Severity.MODERATE);
        logger.warn("Mismatch in the number of files and their corresponding sizes for " + MAPREDUCE_JOB_CACHE_FILES);
        result.addResultDetail(MAPREDUCE_JOB_CACHE_FILES, Integer.toString(cacheFilesCount));
        result.addResultDetail(MAPREDUCE_JOB_CACHE_FILES_FILESIZES, Integer.toString(cacheFileSizesCount));
        return result;
      }

      Map<String, String> cacheFileToSizeMap = new HashMap<String, String>();
      for (int i = 0; i < cacheFilesCount; i++) {
        cacheFileToSizeMap.put(cacheFilesList.get(i), cacheFileSizesList.get(i));
      }

      if (checkFileSizeLimit(result, cacheFileToSizeMap)) {
        result.setSeverity(Severity.CRITICAL);
      }
    }

    String archiveCacheFiles = jobConf.getProperty(MAPREDUCE_JOB_CACHE_ARCHIVES, null);
    String archiveCacheFileSizes = jobConf.getProperty(MAPREDUCE_JOB_CACHE_ARCHIVES_FILESIZES, null);

    if (archiveCacheFiles != null && archiveCacheFileSizes != null) {

      if (result == null) {
        result =
            new HeuristicResult(_heuristicConfData.getClassName(), _heuristicConfData.getHeuristicName(), Severity.NONE,
                0);
      }

      List<String> archiveCacheFilesList = new ArrayList<String>(Arrays.asList(archiveCacheFiles.split(",")));
      List<String> archiveCacheFileSizesList = new ArrayList<String>(Arrays.asList(archiveCacheFileSizes.split(",")));

      int archiveCacheFilesCount = archiveCacheFilesList.size();
      int archiveCacheFileSizesCount = archiveCacheFileSizesList.size();

      if (archiveCacheFilesCount != archiveCacheFileSizesCount) {
        result.setSeverity(Severity.MODERATE);
        logger
            .warn("Mismatch in the number of files and their corresponding sizes for " + MAPREDUCE_JOB_CACHE_ARCHIVES);
        result.addResultDetail(MAPREDUCE_JOB_CACHE_ARCHIVES, Integer.toString(archiveCacheFilesCount));
        result.addResultDetail(MAPREDUCE_JOB_CACHE_ARCHIVES_FILESIZES, Integer.toString(archiveCacheFileSizesCount));
        return result;
      }

      Map<String, String> archiveCacheFileToSizeMap = new HashMap<String, String>();
      for (int i = 0; i < archiveCacheFilesCount; i++) {
        archiveCacheFileToSizeMap.put(archiveCacheFilesList.get(i), archiveCacheFileSizesList.get(i));
      }

      if (checkFileSizeLimit(result, archiveCacheFileToSizeMap)) {
        result.setSeverity(Severity.CRITICAL);
      }
    }

    return result;
  }
Rule flags jobs which put files more than 500MB in the distributed cache.
app/com/linkedin/drelephant/mapreduce/heuristics/DistributedCacheLimitHeuristic.java