g u t x .com.tr İpek yolu - Çin'i anlamaya götürürüm

Hadoop dosya sıkıştırma savaşı (komut dosyaları ve kaynak kodu dahil)

Son zamanlarda, Hadoop küme makinelerinin sabit disk kaynaklarının yetersizliği nedeniyle, Hadoop kümesindeki geçmiş verileri sıkıştırmak gerekiyor (geçmiş veriler sıkıştırılmamış biçimdedir), bu nedenle ilgili eğiticiler olup olmadığını görmek için İnternet'te arama yapmaya başladım. Sonuçların hepsi hakkında Bu sıkıştırma mekanizmasının karşılaştırması, nasıl sıkıştırılacağına dair çok az öğretici var (belki bulamadım ...), işte sıkıştırma işleminin ileride başvurmak için uygun olan ve başkalarına fayda sağlayan bir kaydı.

Bu makaledeki tüm jar paketleri, komut dosyaları ve yerel kitaplık için, makalenin sonundaki ilgili indirmelere bakın ~

Sıkıştırılmış sürümüm:

Jdk 1.7 ve üstü

Hadoop-2.2.0 sürümü

Sıkıştırmadan önce ortam hazırlığı:

Sıkıştırma algoritmalarının karşılaştırılması ile ilgili olarak, İnternette çok fazla bilgi var.Burada daha mütevazı olan Bzip2 sıkıştırma yöntemini kullanıyorum.Hadoop ile birlikte gelen sıkıştırma mekanizması olduğu için başka şeyler indirmeye gerek yok, Hadoop kök dizininde sadece lib / native. Aşağıdaki dosyalar yeterlidir:

Sıkıştırmadan önce, Hadoop kümesi tarafından desteklenen sıkıştırma algoritmasını kontrol edin: hadoop checknative

Her makine kontrol edilmelidir, true şeklinde gösterildiği gibi gösteriliyorsa, bu, kümenin bzip2 sıkıştırmasını desteklediği anlamına gelir.

Yanlış gösteriyorsa, yukarıdaki resimdeki dosyayı indirmeniz ve lib / native Hadoop kök dizinine kopyalamanız gerekir.

Kompresör tanıtımı:

Compressor, getFileList (dosya yolunu elde etmek için), FileHdfsCompress (sıkıştırma sınıfı), FileHdfsDeCompress (açma sınıfı) sınıflarını kullanır. Sıkıştırma / açma işlemini tamamlamak için yalnızca bu üç sınıf kullanılabilir.

getFileList işlevi: Yinelemeli yazdırma Alt dizinlerdeki dosyalar dahil olmak üzere, dosya dizinindeki dosyaların kök yolunu iletin. İlk başta doğrudan bir dosyaya çıktı vermek istedim, daha sonra paketleyip kümeye koyduğumda dosyanın içeriği olmadığını fark ettim. Dağıtılmış işlemden kaynaklanıyor olabilir, bu yüzden yolu yazdırdım ve manuel olarak dosyaya koydum.

Çekirdek kod:

public static void listAllFiles (Dize yolu, Liste < Dize > strings) IOException {FileSystem hdfs = getHdfs (yol); Yol filesAndDirs = getFilesAndDirs (yol); için (Yol p: filesAndDirs) {if (hdfs.getFileStatus (p) .isFile ()) {if (! p.getName ( ) .contains ("SUCCESS")) {System.out.println (p);}} else {listAllFiles (p.toString (), strings);}} // FileUtils.writeLines (yeni Dosya (FILE_LIST_OUTPUT_PATH), dizeler, true);} public static FileSystem getHdfs (String path), IOException oluşturur {Configuration conf = new Configuration (); return FileSystem.get (URI.create (path), conf);} public static Path getFilesAndDirs (String path), IOException { FileStatus fs = getHdfs (yol) .listStatus (yeni Yol (yol)); FileUtil.stat2Paths (fs);} döndür

FileHdfsCompress : Sıkıştırma programı, Hadoop yerel API'sini kullanan, Hadoop kümesindeki orijinal dosyaları giriş akışı olarak okur, çıktı olarak sıkıştırma yolunun giriş akışını alır ve ardından ilgili dosyayı kullanan programdaki FileHdfsCompress sınıfına karşılık gelen çok basittir (FileHdfsDeCompress) Sıkıştırma algoritması, kod aşağıdaki gibidir:

Çekirdek kodu: // Sıkıştırma yöntemi Sınıfını belirtin < ? > codecClass = Class.forName (COMPRESS_CLASS_NAME); Yapılandırma conf = new Configuration (); CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance (codecClass, conf); // FileSystem fs = FileSystem.get (conf); FileSystem fs = FileSystem.get (URI.create (inputPath), conf); // Orijinal dosya yolu txt, = fs.open (yeni Yol (inputPath)) içindeki orijinal giriş akışı ile FSDataInputStream'e okunur; // HDFS'de bir çıktı akışı oluşturun, yolu sıkıştırın / / Çıkış akışını dosya sistemi üzerinden elde edin OutputStream out = fs.create (new Path (FILE_OUTPUT_PATH)); // Çıkış akışının veri sıkıştırması CompressionOutputStream compressOut = codec.createOutputStream (out); // Orijinal dosyayı okuyun ve girdi için HDFS'ye sıkıştırın - Sıradan akış çıktısı sıkıştırılmış akış IOUtils.copyBytes (in, compressOut, 4096, true);

Aşağıdakiler, sıkıştırma programlarının kullanımını içermeyen kod optimizasyon sürecidir ve ilgilenmeyen öğrenciler şunları atlayabilir:

Gerçek kodlamada, aslında bir yol aldım. Başlangıçta, sıkıştırma ve açma işlevini gerçekleştirmek için Hadoop API'yi kullanmayı beklemiyordum. Kod aslında optimizasyon ve yineleme sürecinden geçti.

Başlangıçta sıkıştırma fikri, dosyayı içeri okumak ve sonra sıkıştırmaktı.Başlangıçta, MapReduce kullanıldı.Kodlama işlemi sırasında, sıkıştırılmış dosyayı oluşturma yolunun gereksinimleri nedeniyle, çıktı sınıfı Hadoop çıktısı sırasında özelleştirildi. Çıktı dosyasının adının, part-r-0000.txt değil, timestamp.txt biçimindeki gereksinimleri karşılamasını sağlamak için, şimdiye kadar orijinal çevrimiçi yolun gereksinimlerini karşılar.

Gerçek çalışma sürecinde MR programının İpliği başlatması ve İplik kaynaklarını işgal etmesi gerektiği, uzun sıkıştırma süresinden dolayı küme kaynaklarını serbest bırakmadan uzun süre işgal edebileceği, daha sonra MR programının asıl amacının paralel hesaplama olduğu keşfedildi. Sıkıştırma sadece birini okuyup yazan bir harita görevidir, hesaplamaları içermez, sadece içeriği hareket ettirir. Bu yüzden MR'ı kullanmayı bıraktım ve sıkıştırma işlevini tamamlamak için basit Hadoop API kullanmanın mümkün olup olmadığını düşündüm.Bazı denemelerden sonra bunun gerçekten mümkün olduğunu anladım! Hadoop API, küme kaynaklarını serbest bırakmak için kullanılır ve sıkıştırma hızı da iyidir.Bu şekilde, sıkıştırma programı küme kaynak tahsisi sorunu dikkate alınmadan bir arka plan işlemi olarak çalıştırılabilir. .

Ölçülen sıkıştırma adımları:

1 Projeyi paketleyin ve hadoop kümesinin herhangi bir düğümüne yükleyin.İlgili betiği hazırlayın, aşağıdaki şekilde gösterildiği gibi veri dosyasını ve günlük dosyasını girin:

2 Yolu yazdırmak için Dosya Yolunu Al komut dosyasını kullanın:

GetFileLish.sh dosyasının script içeriği aşağıdaki gibidir, basit bir çağrıdır, Giriş parametresi, hadoop kümesindeki HDFS üzerindeki dizin yoludur :

#! / bin / shecho "begin get fileList" echo "İlk parametre $ 1" if ; thenecho "paramı kontrol et!" fi # orijinal filehadoop jar hadoop-compress-1.0.jar com.people .util.getFileList $ 1

3 Yazdırılan yolu sıkıştır.txt'ye yapıştırın 2. adımda, alt dizin yolunu içeren dizinin dosya yolu yazdırılacak ve daha sonra bunu sıkıştır.txt'ye yapıştıracaktır. Dosya adının serbestçe ayarlanabileceğini unutmayın .

4 Sadece sıkıştırma komut dosyasını kullanın, sh compress.sh /data/new_compress/compress.txt , Kalın yazılmış kısım, komut dosyasının parametresidir, 3. adımdaki dosyanın yolu anlamına gelir. Not: Bu yalnızca mutlak bir yol olabilir, aksi takdirde dosyanın bulunamamasıyla ilgili bir istisna rapor edilebilir.

Compress.sh betiğinin içeriği aşağıdaki gibidir, bu basit bir çağrıdır ve input parametresi 3. adımdaki dosyanın mutlak yoludur:

#! / bin / shecho "sıkıştırmaya başla" echo "İlk parametre $ 1" if ; thenecho "param'ı kontrol et!" fihadoop jar hadoop-compress-1.0.jar com.people.compress.FileHdfsCompress 1 $ > > /data/new_compress/compress.log 2 > 1

5 Sıkıştırma günlüğünü kontrol edin ve arka plan programının sıkıştırmaya başladığını bulun! , Tail -f sıkıştır.log

6 Sıkıştırma hızının yeterli olmadığını düşünüyorsanız, komut dosyasını birden çok makinede çalıştırabilir veya bir makinede birden çok görevi yürütebilirsiniz, çünkü bu komut dosyası görevi bir arka plan işlemidir ve küme Yarn kaynaklarını işgal etmeyecektir.

7 Açma programı, sıkıştırma programına benzer.

İlgili indirmeler:

Nasıl alınır: