Transfer Data Antara HDFS dan OSS dalam Skala Besar dengan Jindo DistCp - E-MapReduce

Jindo DistCp adalah alat penyalinan data terdistribusi yang dibangun untuk E-MapReduce (EMR). Alat ini mentransfer file antara Hadoop Distributed File System (HDFS) dan OSS, serta mendukung penyaringan, kompresi on-the-fly, penggabungan file kecil, penyalinan inkremental, dan integritas transaksi tingkat job.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

Kluster EMR versi 3.28.0 atau lebih baru. Untuk informasi selengkapnya, lihat Create a cluster.
Akses SSH ke node master kluster. Untuk informasi selengkapnya, lihat Log on to a cluster.

Panduan cepat

jindo distcp --src /opt/tmp --dest oss://<yourBucketName>/tmp

Ganti <yourBucketName> dengan nama bucket OSS Anda. Secara default, semua file di bawah --src akan disalin. Jika direktori root tujuan tidak ada, Jindo DistCp akan membuatnya secara otomatis.

Untuk melihat semua opsi yang tersedia, jalankan:

jindo distcp --help

Referensi opsi

Tabel berikut mencantumkan semua opsi yang didukung. Semua contoh dalam dokumen ini menggunakan /data/incoming/hourly_table sebagai sumber dan oss://<yourBucketName>/hourly_table sebagai tujuan. Ganti placeholder tersebut dengan nilai aktual Anda.

Opsi	Deskripsi	Catatan
`--src`	Direktori sumber tempat file dicopy dari	Wajib
`--dest`	Direktori tujuan tempat file dicopy ke	Wajib
`--parallelism`	Jumlah task reduce paralel. Mengatur `mapreduce.job.reduces`. Default: `7`	Sesuaikan berdasarkan sumber daya kluster yang tersedia
`--srcPattern`	Ekspresi reguler untuk menyaring file sumber berdasarkan path lengkap	—
`--deleteOnSuccess`	Menghapus file sumber setelah copy berhasil	Tidak dapat digunakan bersama `--groupBy` atau `--targetSize`
`--outputCodec`	Kodek kompresi untuk file yang dicopy. Nilai yang valid: `gzip`, `gz`, `lzo`, `lzop`, `snappy`, `none`, `keep`. Default: `keep`	`none` mendekompresi file yang sudah dikompresi; `keep` mempertahankan format aslinya
`--outputManifest`	Nama file manifest output (format GZ saja)	Gunakan dengan `--requirePreviousManifest=false` untuk membuat manifest baru
`--requirePreviousManifest`	Mewajibkan adanya manifest sebelumnya sebelum menjalankan	Atur ke `false` saat membuat manifest pertama kali
`--previousManifest`	Path ke file manifest yang sudah ada	Gunakan dengan `--outputManifest` untuk pelacakan copy inkremental
`--copyFromManifest`	Copy hanya file yang tercantum dalam file manifest	Gunakan dengan `--previousManifest` untuk memutar ulang manifest
`--srcPrefixesFile`	Path ke file teks yang berisi satu awalan URI sumber per baris	Memungkinkan copy dari beberapa folder sumber dalam satu job
`--groupBy`	Ekspresi reguler untuk mengelompokkan file input guna penggabungan	Gunakan dengan `--targetSize`
`--targetSize`	Ukuran target untuk file output hasil penggabungan, dalam MB	Gunakan dengan `--groupBy`
`--enableBalancePlan`	Mengoptimalkan alokasi job ketika ukuran file dalam tiap tier (kecil/besar) relatif seragam	Tidak dapat digunakan bersama `--groupBy` atau `--targetSize`
`--enableDynamicPlan`	Mengalokasikan task copy secara dinamis untuk workload dengan variasi ukuran file yang signifikan	Tidak dapat digunakan bersama `--groupBy` atau `--targetSize`
`--enableTransaction`	Mengaktifkan integritas transaksi tingkat job	—
`--diff`	Membandingkan daftar file sumber dan tujuan setelah copy	Jika terdapat perbedaan, menghasilkan file manifest di direktori tujuan; tidak memberikan perbandingan ukuran akurat ketika kompresi atau dekompresi diterapkan
`--cleanUpPending`	Menghapus file sisa dari upload yang gagal, dilacak berdasarkan uploadId	—
`--key`	ID AccessKey untuk akses OSS di luar EMR	Gunakan ketika akses tanpa AccessKey tidak tersedia
`--secret`	Rahasia AccessKey untuk akses OSS di luar EMR	Gunakan ketika akses tanpa AccessKey tidak tersedia
`--endPoint`	Titik akhir OSS	Gunakan ketika akses tanpa AccessKey tidak tersedia
`--policy`	Kelas penyimpanan OSS untuk tujuan. Nilai yang valid: `archive`, `ia`	—

Salin file

Gunakan --src dan --dest untuk menentukan direktori sumber dan tujuan. Secara default, semua file di bawah --src akan disalin. Jika direktori root tujuan tidak ada, Jindo DistCp akan membuatnya secara otomatis.

Perintah berikut menyalin semua file dari direktori HDFS /opt/tmp ke bucket OSS:

jindo distcp --src /opt/tmp --dest oss://<yourBucketName>/tmp

Menyaring file berdasarkan pola

Gunakan --srcPattern untuk hanya menyalin file yang path lengkapnya cocok dengan ekspresi reguler.

Direktori /data/incoming/hourly_table/2017-02-01/03 berisi file-file berikut:

Found 6 items
-rw-r-----   2 root hadoop       2252 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/000151.sst
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/1.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/2.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/OPTIONS-000109
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp01.txt
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp06.txt

Untuk hanya menyalin file .log, atur --srcPattern menjadi .*\.log:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --srcPattern .*\.log --parallelism 20

Verifikasi hasilnya:

hdfs dfs -ls oss://<yourBucketName>/hourly_table/2017-02-01/03

Hanya file log yang disalin:

Found 2 items
-rw-rw-rw-   1       4891 2020-04-17 20:52 oss://<yourBucketName>/hourly_table/2017-02-01/03/1.log
-rw-rw-rw-   1       4891 2020-04-17 20:52 oss://<yourBucketName>/hourly_table/2017-02-01/03/2.log

Menghapus file sumber setelah copy

Tambahkan --deleteOnSuccess untuk menghapus file sumber dari direktori sumber setelah penyalinan berhasil:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --deleteOnSuccess --parallelism 20

Mengompresi file saat copy

Gunakan --outputCodec untuk mengompresi file yang disalin secara langsung. Perintah berikut mengompresi semua file menggunakan gzip:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --outputCodec=gz --parallelism 20

Verifikasi hasilnya:

hdfs dfs -ls oss://<yourBucketName>/hourly_table/2017-02-01/03

Semua file di direktori tujuan memiliki ekstensi .gz:

Found 6 items
-rw-rw-rw-   1        938 2020-04-17 20:58 oss://<yourBucketName>/hourly_table/2017-02-01/03/000151.sst.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://<yourBucketName>/hourly_table/2017-02-01/03/1.log.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://<yourBucketName>/hourly_table/2017-02-01/03/2.log.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://<yourBucketName>/hourly_table/2017-02-01/03/OPTIONS-000109.gz
-rw-rw-rw-   1        506 2020-04-17 20:58 oss://<yourBucketName>/hourly_table/2017-02-01/03/emp01.txt.gz
-rw-rw-rw-   1        506 2020-04-17 20:58 oss://<yourBucketName>/hourly_table/2017-02-01/03/emp06.txt.gz

Nilai yang valid untuk --outputCodec: gzip, gz, lzo, lzop, snappy, none, keep. Default: keep.

none: mendekompresi file jika sudah dikompresi; membiarkan file yang tidak dikompresi tetap seperti aslinya.
keep: mempertahankan format kompresi aslinya.

Untuk menggunakan kodek LZO di kluster Hadoop open-source, instal library gplcompression native dan paket hadoop-lzo.

Melacak copy dengan file manifest

File manifest mencatat path tujuan, path sumber, dan ukuran setiap file yang disalin dalam suatu job. Gunakan file manifest untuk melacak penyalinan inkremental dan memverifikasi kelengkapan.

Buat file manifest

Atur --requirePreviousManifest=false untuk menghasilkan file manifest baru. File tersebut selalu disimpan dalam format GZ.

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --outputManifest=manifest-2020-04-17.gz --requirePreviousManifest=false --parallelism 20

Periksa isi manifest:

hadoop fs -text oss://<yourBucketName>/hourly_table/manifest-2020-04-17.gz > before.lst
cat before.lst

Setiap baris dalam manifest adalah catatan JSON:

{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/000151.sst","baseName":"2017-02-01/03/000151.sst","srcDir":"oss://<yourBucketName>/hourly_table","size":2252}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/1.log","baseName":"2017-02-01/03/1.log","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/2.log","baseName":"2017-02-01/03/2.log","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/OPTIONS-000109","baseName":"2017-02-01/03/OPTIONS-000109","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/emp01.txt","baseName":"2017-02-01/03/emp01.txt","srcDir":"oss://<yourBucketName>/hourly_table","size":1016}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/emp06.txt","baseName":"2017-02-01/03/emp06.txt","srcDir":"oss://<yourBucketName>/hourly_table","size":1016}

Copy hanya file baru menggunakan manifest sebelumnya

Saat file baru ditambahkan ke direktori sumber, gunakan --previousManifest untuk melewati file yang sudah disalin dan hanya mentransfer file baru. Manifest baru (--outputManifest) mencakup file yang sebelumnya dan yang baru disalin, sehingga memberikan riwayat lengkap.

Perintah berikut hanya menyalin file yang ditambahkan sejak manifest-2020-04-17.gz:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --outputManifest=manifest-2020-04-18.gz --previousManifest=oss://<yourBucketName>/hourly_table/manifest-2020-04-17.gz --parallelism 20

Bandingkan kedua manifest untuk melihat file yang baru disalin:

hadoop fs -text oss://<yourBucketName>/hourly_table/manifest-2020-04-18.gz > current.lst
diff before.lst current.lst

Output diff menunjukkan dua file baru yang ditambahkan:

3a4,5
> {"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/5.log","baseName":"2017-02-01/03/5.log","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
> {"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/6.log","baseName":"2017-02-01/03/6.log","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}

Copy file dari manifest

Gunakan --copyFromManifest dengan --previousManifest untuk menyalin persis file yang tercantum dalam manifest yang sebelumnya dihasilkan:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --previousManifest=oss://<yourBucketName>/hourly_table/manifest-2020-04-17.gz --copyFromManifest --parallelism 20

Copy file dari beberapa folder sumber

--srcPrefixesFile menerima file teks yang berisi satu awalan URI sumber per baris, memungkinkan satu job menyalin dari beberapa folder.

Daftar folder sumber:

hdfs dfs -ls oss://<yourBucketName>/hourly_table

Found 4 items
drwxrwxrwx   -          0 1970-01-01 08:00 oss://<yourBucketName>/hourly_table/2017-02-01
drwxrwxrwx   -          0 1970-01-01 08:00 oss://<yourBucketName>/hourly_table/2017-02-02

Jalankan penyalinan dan tentukan file prefixes:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --srcPrefixesFile file:///opt/folders.txt --parallelism 20

Lihat isi folders.txt yang dihasilkan untuk mengonfirmasi isinya:

cat folders.txt

hdfs://emr-header-1.cluster-50466:9000/data/incoming/hourly_table/2017-02-01
hdfs://emr-header-1.cluster-50466:9000/data/incoming/hourly_table/2017-02-02

Menggabungkan file kecil

Membaca banyak file kecil dari HDFS menurunkan performa pemrosesan. Gunakan --groupBy dan --targetSize secara bersamaan untuk menggabungkan file kecil menjadi file output yang lebih besar. --groupBy adalah ekspresi reguler yang mengelompokkan file; --targetSize menetapkan ukuran maksimum tiap file gabungan dalam MB.

Direktori sumber berisi delapan file kecil:

hdfs dfs -ls /data/incoming/hourly_table/2017-02-01/03

Found 8 items
-rw-r-----   2 root hadoop       2252 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/000151.sst
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/1.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/2.log
-rw-r-----   2 root hadoop       4891 2020-04-17 21:08 /data/incoming/hourly_table/2017-02-01/03/5.log
-rw-r-----   2 root hadoop       4891 2020-04-17 21:08 /data/incoming/hourly_table/2017-02-01/03/6.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/OPTIONS-000109
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp01.txt
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp06.txt

Perintah berikut menggabungkan semua file .txt menjadi satu file output dengan ukuran maksimal 10 MB:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --targetSize=10 --groupBy='.*/([a-z]+).*.txt' --parallelism 20

Verifikasi hasilnya. Dua file TXT digabung menjadi satu:

hdfs dfs -ls oss://<yourBucketName>/hourly_table/2017-02-01/03/
Found 1 items
-rw-rw-rw-   1       2032 2020-04-17 21:18 oss://<yourBucketName>/hourly_table/2017-02-01/03/emp2

Mengoptimalkan performa copy

Jindo DistCp menyediakan dua strategi alokasi job untuk meningkatkan performa saat menyalin file dengan ukuran campuran. Pilih berdasarkan workload Anda, lalu sesuaikan --parallelism untuk mengontrol konkurensi.

Strategi	Kapan digunakan	Opsi
Balance plan	Ukuran file dalam kelompok file kecil dan dalam kelompok file besar relatif seragam	`--enableBalancePlan`
Dynamic plan	Ukuran file sangat bervariasi antar file, dengan sebagian besar file berukuran kecil	`--enableDynamicPlan`

Jika tidak ada opsi yang ditentukan, file dialokasikan secara acak ke task reduce.

Opsi --enableBalancePlan maupun --enableDynamicPlan tidak dapat digunakan bersama --groupBy atau --targetSize.

Balance plan:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --enableBalancePlan --parallelism 20

Dynamic plan:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --enableDynamicPlan --parallelism 20

Sesuaikan --parallelism untuk mengontrol jumlah task reduce konkuren. Nilai default adalah 7. Tingkatkan nilai ini ketika kluster memiliki lebih banyak sumber daya yang tersedia untuk meningkatkan throughput penyalinan.

Aktifkan integritas transaksi

--enableTransaction memberlakukan integritas transaksi tingkat job. Semua file dalam job tersebut akan berhasil atau dikembalikan (rollback) secara bersamaan:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --enableTransaction --parallelism 20

Verifikasi kelengkapan copy

Setelah job penyalinan selesai, gunakan --diff untuk membandingkan daftar file sumber dan tujuan:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --diff

Jika semua file berhasil disalin:

INFO distcp.JindoDistCp: distcp has been done completely

Jika ada file yang hilang dari tujuan, Jindo DistCp menghasilkan file manifest di direktori tujuan yang mencantumkannya. Gunakan --copyFromManifest dan --previousManifest untuk menyalin file tersebut:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --previousManifest=file:///opt/manifest-2020-04-17.gz --copyFromManifest --parallelism 20

Jika Jindo DistCp melakukan kompresi atau dekompresi selama penyalinan, --diff tidak memberikan perbandingan ukuran file yang akurat. Jika tujuan adalah direktori HDFS, tentukan --dest dalam salah satu format berikut: /path, hdfs://hostname:ip/path, atau hdfs://headerIp:ip/path. Format lain seperti hdfs:///path dan hdfs:/path tidak didukung.

Periksa counter DistCp

Setelah job selesai, periksa bagian DistCp Counters dalam output job MapReduce untuk memverifikasi berapa banyak data yang ditransfer:

Distcp Counters
        Bytes Destination Copied=11010048000
        Bytes Source Read=11010048000
        Files Copied=1001

Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0

Jika Jindo DistCp mengompresi atau mendekompresi file selama penyalinan, nilai Bytes Destination Copied dan Bytes Source Read mungkin berbeda.

Akses OSS dengan pasangan AccessKey

Secara default, EMR menggunakan akses tanpa AccessKey untuk OSS. Jika Anda menjalankan Jindo DistCp dari luar EMR atau akses tanpa AccessKey tidak tersedia, tentukan pasangan AccessKey menggunakan --key, --secret, dan --endPoint:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --key <yourAccessKeyId> --secret <yourAccessKeySecret> --endPoint oss-cn-hangzhou.aliyuncs.com --parallelism 20

Ganti <yourAccessKeyId> dengan ID AccessKey Alibaba Cloud Anda dan <yourAccessKeySecret> dengan rahasia AccessKey Anda.

Menulis ke penyimpanan OSS Archive atau Infrequent Access

Gunakan --policy untuk menulis data langsung ke kelas penyimpanan OSS berbiaya lebih rendah selama penyalinan.

Penyimpanan Archive:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --policy archive --parallelism 20

Penyimpanan Infrequent Access (IA):

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --policy ia --parallelism 20

Bersihkan file sisa

Ketika task DistCp gagal di tengah jalan, file yang diupload sebagian mungkin tersisa di tujuan. OSS melacak file-file ini secara internal berdasarkan uploadId, tetapi file tersebut tidak terlihat dalam daftar direktori. Tambahkan --cleanUpPending untuk membersihkannya secara otomatis setelah task selesai:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --cleanUpPending --parallelism 20

Atau, hapus file sisa secara manual dari Konsol OSS.

FAQ

Mengapa `--diff` melaporkan ukuran file salah setelah kompresi?

Ketika Jindo DistCp mengompresi atau mendekompresi file selama penyalinan, ukuran file sumber dan tujuan memang berbeda secara desain. --diff membandingkan ukuran file mentah sehingga melaporkan ketidaksesuaian. Ini adalah perilaku yang diharapkan, bukan kegagalan penyalinan. Periksa DistCp Counters sebagai gantinya untuk mengonfirmasi jumlah file yang disalin.

Apakah saya bisa menggunakan `--enableBalancePlan` atau `--enableDynamicPlan` dengan `--groupBy`?

Tidak. Kedua opsi rencana tersebut tidak kompatibel dengan --groupBy dan --targetSize. Saat menggabungkan file kecil, alokasi file ditentukan oleh pola pengelompokan, sehingga opsi rencana tidak berpengaruh dan tidak diizinkan dalam perintah yang sama.

Apakah saya bisa menggunakan `--deleteOnSuccess` dengan `--groupBy`?

Tidak. --deleteOnSuccess tidak dapat digunakan dengan --groupBy atau --targetSize. Setelah penggabungan, outputnya adalah file gabungan baru; menghapus input asli dalam job yang sama tidak didukung.

Mengapa format path HDFS tujuan dibatasi untuk `--diff`?

Ketika tujuan adalah direktori HDFS, Jindo DistCp mewajibkan --dest dalam format /path, hdfs://hostname:ip/path, atau hdfs://headerIp:ip/path. Format seperti hdfs:///path dan hdfs:/path tidak didukung dan akan menyebabkan operasi diff gagal.