Gunakan Spark untuk membaca data dari OSS - E-MapReduce

Dengan mengintegrasikan Spark dan Object Storage Service (OSS), Alibaba Cloud E-MapReduce (EMR) memungkinkan pemrosesan serta analisis data yang efisien di data lake berbasis cloud. EMR memungkinkan Anda membaca dan menulis data ke OSS tanpa perlu menentukan pasangan AccessKey atau dengan menentukannya secara eksplisit. Topik ini menjelaskan cara menggunakan Spark untuk memproses dan menganalisis data OSS.

Baca data dari dan tulis data ke OSS tanpa menentukan pasangan AccessKey

Gunakan Resilient Distributed Dataset (RDD) Spark untuk membaca data dari dan menulis data ke OSS

Contoh berikut menunjukkan cara menggunakan Spark untuk membaca data dari OSS dan menulis data yang telah diproses kembali ke OSS tanpa menentukan pasangan AccessKey:

Masuk ke node master kluster Anda dalam mode SSH. Untuk informasi lebih lanjut, lihat Masuk ke Node Master Kluster.
Jalankan perintah berikut untuk memulai Spark Shell:
```
spark-shell
```

Setelah memodifikasi parameter dalam kode berikut sesuai kebutuhan bisnis Anda, jalankan kode Scala berikut di Spark Shell untuk membaca dan menulis data ke OSS.

import org.apache.spark.{SparkConf, SparkContext}
val conf = new SparkConf().setAppName("Test OSS")
val sc = new SparkContext(conf)
val pathIn = "oss://<yourBucket>/path/to/read"
val inputData = sc.textFile(pathIn)
val cnt = inputData.count
println(s"count: $cnt")
val outputPath = "oss://<yourBucket>/path/to/write"
val outputData = inputData.map(e => s"$e telah diproses.")
outputData.saveAsTextFile(outputPath)

Dalam contoh ini, ganti parameter berikut sesuai kebutuhan bisnis Anda:

yourBucket: Nama Bucket OSS.
pathIn: Jalur file dari mana Anda ingin membaca data.
outputPath: Jalur file ke mana Anda ingin menulis data.

Untuk kode sampel lengkap, kunjungi GitHub.

Gunakan PySpark untuk membaca data dari dan menulis data ke OSS

Contoh berikut menunjukkan cara menggunakan PySpark untuk membaca data dari OSS dan menulis data yang telah diproses kembali ke OSS tanpa menentukan pasangan AccessKey:

Masuk ke node master kluster Anda dalam mode SSH. Untuk informasi lebih lanjut, lihat Masuk ke Node Master Kluster.
Jalankan perintah berikut untuk masuk ke lingkungan interaktif PySpark:
```
pyspark
```

Setelah memodifikasi parameter dalam kode berikut sesuai kebutuhan bisnis Anda, jalankan kode tersebut di PySpark.

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Python Spark SQL OSS example").getOrCreate()
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)

Dalam contoh ini, ganti parameter berikut sesuai kebutuhan bisnis Anda:

yourBucket: Nama Bucket OSS.
pathIn: Jalur file dari mana Anda ingin membaca data.
outputPath: Jalur file ke mana Anda ingin menulis data.

Eksekusi pernyataan SQL Spark untuk membuat file CSV dan menyimpan file di OSS

Contoh berikut menunjukkan cara mengeksekusi pernyataan SQL Spark untuk membuat file CSV dan menyimpannya di OSS.

Masuk ke node master kluster Anda dalam mode SSH. Untuk informasi lebih lanjut, lihat Masuk ke Node Master Kluster.
Jalankan perintah berikut untuk membuka CLI Spark SQL:
```
spark-sql
```
Setelah memodifikasi parameter dalam kode berikut sesuai kebutuhan bisnis Anda, jalankan kode tersebut di Spark SQL.
```
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;     
```
Parameter dalam kode:
- yourBucket: Nama Bucket OSS.
- delimiter: Pemisah yang digunakan untuk memisahkan data dalam file CSV.
- header: Tentukan apakah baris pertama dalam file CSV adalah header tabel. Nilai true menunjukkan bahwa baris pertama adalah header tabel. Nilai false menunjukkan sebaliknya.
Setelah pernyataan SQL Spark dieksekusi, hasil berikut dikembalikan:
```
1    ab    12     
```
Lihat file CSV yang disimpan di OSS.
Kode berikut memberikan informasi yang terkandung dalam file CSV. Baris pertama adalah header tabel yang dipisahkan oleh titik koma (;).
```
id;name;age
1;ab;12
```

Baca data dari dan tulis data ke OSS dengan secara eksplisit menentukan pasangan AccessKey

Contoh berikut menunjukkan cara menentukan pasangan AccessKey secara eksplisit untuk kluster EMR agar dapat mengakses OSS.

Hapus konfigurasi bebas kata sandi.
Anda dapat mengakses OSS dari kluster EMR tanpa memasukkan kata sandi. Jika ingin menghapus konfigurasi bebas kata sandi, hapus pengaturan parameter fs.oss.credentials.provider dari file core-site.xml layanan Hadoop-Common.
Jalankan perintah berikut untuk memeriksa apakah akses bebas kata sandi tersedia:
```
hadoop fs -ls oss://<yourBucket>/test_db
```
Informasi berikut dikembalikan, yang menunjukkan bahwa Anda tidak dapat mengakses OSS setelah menghapus pengaturan parameter fs.oss.credentials.provider.
```
ls: ERROR: not found login secrets, please configure the accessKeyId and accessKeySecret.
```
Tentukan pasangan AccessKey secara eksplisit untuk kluster EMR agar dapat mengakses OSS.
Untuk mengakses OSS dengan pasangan AccessKey setelah menghapus konfigurasi bebas kata sandi, tambahkan parameter berikut terkait pasangan AccessKey ke file core-site.xml layanan Hadoop-Common.
Kunci
Nilai
Deskripsi
fs.oss.accessKeyId
LTAI5tM85Z4sc****
ID AccessKey.
fs.oss.accessKeySecret
HF7P1L8PS6Eqf****
Rahasia AccessKey.
Setelah menentukan pasangan AccessKey yang digunakan untuk mengakses OSS, jalankan perintah berikut untuk memeriksa apakah pasangan AccessKey valid:
```
hadoop fs -ls oss://<yourBucket>/test_db
```
Jika informasi berikut dikembalikan, Anda dapat melihat jalur file OSS.
```
drwxrwxrwx   - root root          0 2025-02-24 11:45 oss://<yourBucket>/test_db/student
```
Mulai ulang layanan terkait Spark. Setelah layanan terkait Spark mulai berjalan normal, Anda dapat menggunakan RDD Spark, PySpark, atau Spark SQL untuk membaca dan menulis data ke OSS.

FAQ

Bagaimana cara saya menggunakan Spark untuk membaca data dari Bucket A dan menulis data ke Bucket B?

Jika Bucket A dan Bucket B menggunakan kredensial yang berbeda, Anda dapat mengonfigurasi penyedia kredensial tingkat bucket. Misalnya, konfigurasikan parameter fs.oss.bucket.<BucketName>.credentials.provider. <BucketName> menentukan nama Bucket OSS yang ingin Anda konfigurasikan. Untuk informasi lebih lanjut, lihat Konfigurasikan Penyedia Kredensial OSS atau OSS-HDFS Berdasarkan Bucket.
Jika Bucket A dan Bucket B berada di wilayah yang berbeda, Anda dapat menggunakan oss://<BucketName>.<Public endpoint of the bucket>/ untuk memungkinkan Spark membaca dan menulis data ke OSS. Jika menggunakan metode ini, biaya lalu lintas akan dikenakan dan masalah stabilitas dapat terjadi.

Bagaimana cara saya menggunakan SDK Amazon S3 untuk mengakses OSS?

OSS menyediakan operasi API yang kompatibel dengan Amazon Simple Storage Service (Amazon S3). Setelah memigrasikan data Anda dari Amazon S3 ke OSS, Anda dapat membuat aplikasi klien Anda kompatibel dengan OSS dengan melakukan modifikasi konfigurasi sederhana. Untuk informasi lebih lanjut, lihat Gunakan SDK Amazon S3 untuk Mengakses OSS.

Kunci	Nilai	Deskripsi
fs.oss.accessKeyId	LTAI5tM85Z4sc****	ID AccessKey.
fs.oss.accessKeySecret	HF7P1L8PS6Eqf****	Rahasia AccessKey.