Spark를 사용하여 HDFS에서 Harz 파일을 압축하는 중

Question

hadoop 아카이브 .har 형식의 거대한 데이터가 있습니다. har에는 압축이 포함되어 있지 않으므로 추가로 gzip으로 압축하여 HDFS에 저장하려고합니다. 내가 실수없이 일할 수있는 유일한 것은 :

harFile.coalesce(1, "true") 
.saveAsTextFile("hdfs://namenode/archive/GzipOutput", classOf[org.apache.hadoop.io.compress.GzipCodec]) 
//`coalesce` because Gzip isn't splittable. 

그러나 이것은 올바른 결과를주지 못합니다. Gzipped 파일이 생성되었지만 잘못된 출력 (rdd 유형을 말하는 한 줄 등)

도움이 되겠습니다. 저는 다른 접근법에 대해서도 열려 있습니다.

감사합니다.

Answer 1

기존 HDFS 파일의 압축 버전을 만드는 Java 코드 스 니펫.

서둘러, 텍스트 편집기에서 Java App의 비트와 조각을 사용하여 작성했습니다. 이전에 작성 했으므로 테스트하지 않았습니다. 일부 오타 및 예상되는 간격.

// HDFS API 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.security.UserGroupInformation; 
import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.fs.FileStatus; 
// native Hadoop compression libraries 
import org.apache.hadoop.io.compress.CompressionCodecFactory; 
import org.apache.hadoop.io.compress.CompressionCodec; 
import org.apache.hadoop.io.compress.Compressor; 
import org.apache.hadoop.io.compress.GzipCodec; 
import org.apache.hadoop.io.compress.BZip2Codec; 
import org.apache.hadoop.io.compress.SnappyCodec; 
import org.apache.hadoop.io.compress.Lz4Codec; 

.............. 

    // Hadoop "Configuration" (and its derivatives for HDFS, HBase etc.) constructors try to auto-magically 
    // find their config files by searching CLASSPATH for directories, and searching each dir for hard-coded 
    // name "core-site.xml", plus "hdfs-site.xml" and/or "hbase-site.xml" etc. 
    // WARNING - if these config files are not found, the "Configuration" reverts to hard-coded defaults without 
    // any warning, resulting in bizarre error messages later > let's run some explicit controls here 
    Configuration cnfHadoop = new Configuration() ; 
    String propDefaultFs =cnfHadoop.get("fs.defaultFS") ; 
    if (propDefaultFs ==null || ! propDefaultFs.startsWith("hdfs://")) 
    { throw new IllegalArgumentException(
       "HDFS configuration is missing - no proper \"core-site.xml\" found, please add\n" 
       +"directory /etc/hadoop/conf/ (or custom dir with custom XML conf files) in CLASSPATH" 
       ) ; 
    } 
/* 
    // for a Kerberised cluster, either you already have a valid TGT in the default 
    // ticket cache (via "kinit"), or you have to authenticate by code 
    UserGroupInformation.setConfiguration(cnfHadoop) ; 
    UserGroupInformation.loginUserFromKeytab("user@REALM", "/some/path/to/user.keytab") ; 
*/ 
    FileSystem fsCluster =FileSystem.get(cnfHadoop) ; 
    Path source = new Path("/some/hdfs/path/to/XXX.har") ; 
    Path target = new Path("/some/hdfs/path/to/XXX.har.gz") ; 

    // alternative: "BZip2Codec" for better compression (but higher CPU cost) 
    // alternative: "SnappyCodec" or "Lz4Codec" for lower compression (but much lower CPU cost) 
    CompressionCodecFactory codecBootstrap = new CompressionCodecFactory(cnfHadoop) ; 
    CompressionCodec codecHadoop =codecBootstrap.getCodecByClassName(GzipCodec.class.getName()) ; 
    Compressor compressorHadoop =codecHadoop.createCompressor() ; 

    byte[] buffer = new byte[16*1024*1024] ; 
    int bufUsedCapacity ; 
    InputStream sourceStream =fsCluster.open(source) ; 
    OutputStream targetStream =codecHadoop.createOutputStream(fsCluster.create(target, true), compressorHadoop) ; 
    while ((bufUsedCapacity =sourceStream.read(buffer)) >0) 
    { targetStream.write(buffer, 0, bufUsedCapacity) ; } 
    targetStream.close() ; 
    sourceStream.close() ; 

..............