BaseUploader { public void upload(String clusterName, String fileSet, Options options) throws Exception { List<Pair<Path, Long>> fileSizePairList = getFileList(); int numShards = fileSizePairList.size(); LOG.info("Got " + numShards + " files."); if (numShards == 0) { LOG.warn("No files found. Exiting."); System.exit(1); } List<Path> parts = Lists.transform(fileSizePairList, new Function<Pair<Path, Long>, Path>() { @Override public Path apply(Pair<Path, Long> pathLongPair) { return pathLongPair.getKey(); } }); PartitionerType partitionerType = options.getPartitioner(); validate(parts, partitionerType, numShards); long maxSize = -1; for (Pair<Path, Long> fileSizePair : fileSizePairList) { long size = fileSizePair.getRight(); if (maxSize < size) { maxSize = size; } } long timestampEpochMillis = System.currentTimeMillis(); String hdfsDir = Constants.HDFS_DATA_DIR + "/" + fileSet + "/" + timestampEpochMillis; ZooKeeperManager zkManager = getZKManager(clusterName); FileSetInfo fileSetInfo = new FileSetInfo(fileSet, hdfsDir, numShards, (List)Lists.newArrayList(), options); int replicationFactor = Constants.DEFAULT_HDFS_REPLICATION; if (terrapinNamenode == null || terrapinNamenode.isEmpty()) { ClusterInfo info = zkManager.getClusterInfo(); if (info == null) { LOG.error("Could not find the namenode for " + clusterName); System.exit(1); } if (info.hdfsNameNode == null || info.hdfsNameNode.isEmpty()) { LOG.error("Could not find the namenode for " + clusterName); System.exit(1); } this.terrapinNamenode = info.hdfsNameNode; replicationFactor = info.hdfsReplicationFactor; } LOG.info("Locking fileset " + fileSet); zkManager.lockFileSet(fileSet, fileSetInfo); try { LOG.info("Uploading " + numShards + " files through distcp to " + hdfsDir); List<Path> sourceFiles = Lists.newArrayListWithCapacity(fileSizePairList.size()); for (Pair<Path, Long> fileSize : fileSizePairList) { sourceFiles.add(fileSize.getLeft()); } if (sourceFiles.size() == 1) { hdfsDir = hdfsDir + "/" + TerrapinUtil.formatPartitionName(0); } DistCpOptions distCpOptions = new DistCpOptions(sourceFiles, new Path("hdfs", terrapinNamenode, hdfsDir)); distCpOptions.setSyncFolder(true); distCpOptions.setSkipCRC(true); if (maxSize > Constants.DEFAULT_MAX_SHARD_SIZE_BYTES) { LOG.warn("Largest shard is " + maxSize + " bytes. This is more than 4G. " + "Increase the # of shards to reduce the size."); System.exit(1); } TerrapinUtil.setupConfiguration(conf, maxSize, replicationFactor); DistCp distCp = getDistCp(conf, distCpOptions); Job job = distCp.execute(); if (!job.waitForCompletion(true)) { throw new RuntimeException("Distributed copy failed."); } LOG.info("Successfully copied data."); loadFileSetData(zkManager, fileSetInfo, options); try { LOG.info("Releasing file set lock."); Thread.sleep(5000); } catch (InterruptedException ie) { LOG.warn("Interrupted."); } } finally { zkManager.unlockFileSet(fileSet); } } BaseUploader(TerrapinUploaderOptions uploaderOptions); void validate(List<Path> parts, PartitionerType partitionerType, int numShards); void upload(String clusterName, String fileSet, Options options);  }
@Test public void testUploadOneFile() throws Exception { uploader.init(1); uploader.upload(CLUSTER, FILE_SET, new Options()); uploader.verifyTest(); }
@Test public void testUpload() throws Exception { uploader.init(100); uploader.upload(CLUSTER, FILE_SET, new Options()); uploader.verifyTest(); }