code: public static void createFileBatchesSpark(JavaRDD<String> filePaths, final String rootOutputDir, final int batchSize, @NonNull final org.apache.hadoop.conf.Configuration hadoopConfig) {
          final SerializableHadoopConfig conf = new SerializableHadoopConfig(hadoopConfig);
          //Here: assume input is images. We can't store them as Float32 arrays - that's too inefficient
          // instead: let's store the raw file content in a batch.
          long count = filePaths.count();
          long maxPartitions = count / batchSize;
          JavaRDD<String> repartitioned = filePaths.repartition(Math.max(filePaths.getNumPartitions(), (int) maxPartitions));
          repartitioned.foreachPartition(new VoidFunction<Iterator<String>>() {
              @Override
              public void call(Iterator<String> stringIterator) throws Exception {
                  //Construct file batch
                  List<String> list = new ArrayList<>();
                  List<byte[]> bytes = new ArrayList<>();
                  FileSystem fs = FileSystem.get(conf.getConfiguration());
                  while (stringIterator.hasNext()) {
                      String inFile = stringIterator.next();
                      byte[] fileBytes;
                      try (BufferedInputStream bis = new BufferedInputStream(fs.open(new Path(inFile)))) {
                          fileBytes = IOUtils.toByteArray(bis);
                      }
                      list.add(inFile);
                      bytes.add(fileBytes);
  
                      if (list.size() == batchSize) {
                          process(list, bytes);
                      }
                  }
                  if (list.size() > 0) {
                      process(list, bytes);
                  }
              }
  
              private void process(List<String> paths, List<byte[]> bytes) throws IOException {
                  FileBatch fb = new FileBatch(bytes, paths);
                  String name = UUID.randomUUID().toString().replaceAll("-", "") + ".zip";
                  String outPath = FilenameUtils.concat(rootOutputDir, name);
                  FileSystem fileSystem = FileSystem.get(conf.getConfiguration());
                  try (BufferedOutputStream bos = new BufferedOutputStream(fileSystem.create(new Path(outPath)))) {
                      fb.writeAsZip(bos);
                  }
  
                  paths.clear();
                  bytes.clear();
              }
          });
      }
docs: See {@link #createFileBatchesSpark(JavaRDD, String, int, JavaSparkContext)}