E-MapReduce:Hubungkan Spark ke OSS

Gunakan Spark Shell (Scala)

1. Masuk ke node master melalui SSH.

2. Jalankan Spark Shell:

   spark-shell

3. Jalankan kode berikut. Ganti <yourBucket> dengan nama bucket OSS Anda.

   scala> val pathIn = "oss://<yourBucket>/path/to/read"
   scala> val inputData = sc.textFile(pathIn)
   scala> val cnt = inputData.count
   cnt: Long = ...
   scala> println(s"count: $cnt")
   scala> val outputPath = "oss://<yourBucket>/path/to/write"
   scala> val outputData = inputData.map(e => s"$e has been processed.")
   scala> outputData.saveAsTextFile(outputPath)

   Kode ini membaca semua baris dari path input, menghitung jumlahnya, menambahkan sufiks ke setiap baris, lalu menulis hasilnya ke path output. Untuk contoh lengkapnya, lihat SparkOssDemo.scala di GitHub.

Gunakan PySpark

1. Masuk ke node master melalui SSH.

2. Jalankan PySpark:

   pyspark

3. Jalankan kode berikut. Ganti <yourBucket> dengan nama bucket OSS Anda.

   >>> pathIn = "oss://<yourBucket>/path/to/read"
   >>> df = spark.read.text(pathIn)
   >>> cnt = df.count()
   >>> print(cnt)
   >>> outputPath = "oss://<yourBucket>/path/to/write"
   >>> df.write.format("parquet").mode("overwrite").save(outputPath)

   Kode ini membaca path input sebagai DataFrame teks, mencetak jumlah barisnya, lalu menulis hasilnya dalam format Parquet ke path output.

Gunakan Spark SQL

1. Masuk ke node master melalui SSH.

2. Jalankan CLI Spark SQL:

   spark-sql

3. Buat database yang disimpan di OSS, buat tabel CSV, dan masukkan satu baris:

   Parameter	Deskripsi
   delimiter	Karakter yang digunakan untuk memisahkan bidang dalam file CSV.
   header	Atur ke true jika baris pertama berisi nama kolom; false jika tidak.

   CREATE DATABASE test_db LOCATION "oss://<yourBucket>/test_db";
   USE test_db;
   CREATE TABLE student (id INT, name STRING, age INT)
       USING CSV OPTIONS ("delimiter"=";", "header"="true");
   INSERT INTO student VALUES(1, "ab", 12);
   SELECT * FROM student;

   Ganti <yourBucket> dengan nama bucket OSS Anda. Pernyataan SELECT mengembalikan:

   1    ab    12

4. Untuk memverifikasi hasilnya, periksa file CSV di OSS. File tersebut menggunakan delimiter titik koma, dengan baris pertama sebagai header:

   id;name;age
   1;ab;12

Langkah 1: Hapus konfigurasi akses tanpa password

Hapus parameter fs.oss.credentials.provider dari file core-site.xml layanan Hadoop-Common.

Langkah 2: Verifikasi bahwa akses tanpa password telah dihapus

Jalankan perintah berikut:

hadoop fs -ls oss://<yourBucket>/test_db

Jika penghapusan berhasil, Anda akan melihat:

ls: ERROR: not found login secrets, please configure the accessKeyId and accessKeySecret.

Langkah 3: Tambahkan parameter AccessKey ke core-site.xml

Dalam file core-site.xml layanan Hadoop-Common, tambahkan parameter berikut:

Langkah 4: Verifikasi konfigurasi AccessKey

Jalankan perintah berikut:

hadoop fs -ls oss://<yourBucket>/test_db

Jika pasangan AccessKey dikonfigurasi dengan benar, output akan menampilkan daftar path OSS:

drwxrwxrwx   - root root          0 2025-02-24 11:45 oss://<yourBucket>/test_db/student

Langkah 5: Mulai ulang layanan Spark

Mulai ulang semua layanan terkait Spark. Setelah berjalan, gunakan Spark Shell, PySpark, atau Spark SQL untuk membaca dari dan menulis ke OSS.