Masalah umum dan solusi saat menggunakan Spark - E-MapReduce

Topik ini menjawab pertanyaan umum tentang penggunaan Spark.

Spark Core
Spark SQL
- Bagaimana cara menggunakan fitur penggabungan file kecil di Spark 3?
- Bagaimana cara menangani data skew di Spark SQL?
PySpark
- Bagaimana cara menentukan Python 3 untuk PySpark?
Spark Streaming
- Mengapa pekerjaan Spark Streaming saya berhenti secara tiba-tiba setelah berjalan beberapa saat?
- Mengapa pekerjaan Spark Streaming yang telah dihentikan masih ditampilkan sebagai Running di Konsol EMR?
spark-submit
- Mengapa saya mendapatkan error java.lang.ClassNotFoundException saat menjalankan spark-submit dalam mode YARN-cluster pada kluster EMR yang diaktifkan Kerberos?

Di mana saya dapat melihat pekerjaan Spark historis?

Di Konsol EMR, buka tab Access Links and Ports pada kluster target, lalu klik tautan Spark UI untuk melihat pekerjaan Spark historis. Untuk informasi lebih lanjut tentang mengakses UI komponen, lihat Akses web UI komponen open-source di Konsol EMR.

Dapatkah saya mengirimkan pekerjaan Spark dalam mode standalone?

Tidak. E-MapReduce hanya mendukung pengiriman pekerjaan melalui Spark on YARN dan Spark on Kubernetes. Mode standalone dan Mesos tidak didukung.

Cara mengurangi output log CLI Spark 2

Secara default, alat command-line interface (CLI) seperti spark-sql dan spark-shell pada kluster EMR DataLake menghasilkan log tingkat INFO.

Pada node tempat Anda menjalankan alat CLI (misalnya, node master), buat file konfigurasi log4j.properties. Untuk menyalin file konfigurasi default, gunakan perintah berikut:
```
cp /etc/emr/spark-conf/log4j.properties /new/path/to/log4j.properties
```
Ubah tingkat log dalam file konfigurasi baru tersebut.
```
log4j.rootCategory=WARN, console
```
Pada file spark-defaults.conf untuk layanan Spark, perbarui properti spark.driver.extraJavaOptions. Ganti -Dlog4j.configuration=file:/etc/emr/spark-conf/log4j.properties dengan -Dlog4j.configuration=file:/new/path/to/log4j.properties.

Penting
Jalur harus diawali dengan file:.

Cara menggunakan penggabungan file kecil Spark 3

Atur parameter spark.sql.adaptive.merge.output.small.files.enabled ke true untuk secara otomatis menggabungkan file kecil. File yang digabung dikompresi. Jika file hasil penggabungan terlalu kecil, tingkatkan nilai parameter spark.sql.adaptive.advisoryOutputFileSizeInBytes. Nilai default-nya adalah 256 MB.

Cara menangani data skew di Spark SQL

Untuk Spark 2, gunakan salah satu pendekatan berikut:

Filter data yang tidak relevan, seperti nilai null, saat membaca tabel.

Broadcast tabel yang lebih kecil.

select /*+ BROADCAST (table1) */ * from table1 join table2 on table1.id = table2.id

Pisahkan data yang miring berdasarkan kunci yang miring.

select * from table1_1 join table2 on table1_1.id = table2.id
union all
select /*+ BROADCAST (table1_2) */ * from table1_2 join table2 on table1_2.id = table2.id

Jika kunci yang miring diketahui, sebarkan datanya.

select id, value, concat(id, (rand() * 10000) % 3) as new_id from A
select id, value, concat(id, suffix) as new_id
from ( 
select id, value, suffix
 from B Lateral View explode(array(0, 1, 2)) tmp as suffix)

Jika kunci yang miring tidak diketahui, sebarkan datanya.

select t1.id, t1.id_rand, t2.name
from (
select id ,
case when id = null then concat('SkewData_', cast(rand() as string))
else id end as id_rand
from test1
where statis_date = '20221130') t1
left join test2 t2
on t1.id_rand = t2.id

Untuk Spark 3, buka tab Configure layanan Spark 3 di Konsol EMR, lalu atur parameter spark.sql.adaptive.enabled dan spark.sql.adaptive.skewJoin.enabled ke true.

Cara menentukan Python 3 untuk PySpark

Bagian ini menjelaskan cara menentukan Python 3 untuk PySpark, menggunakan kluster DataLake pada EMR V5.7.0 dengan Spark 2 sebagai contoh.

Ada dua metode untuk mengubah versi Python:

Metode sementara

Login ke kluster melalui SSH. Untuk informasi lebih lanjut, lihat Login ke kluster.
Jalankan perintah berikut untuk mengubah versi Python:
```
export PYSPARK_PYTHON=/usr/bin/python3
```
Jalankan perintah berikut untuk memeriksa versi Python:
```
pyspark
```
Jika output berisi pesan berikut, versi Python telah diubah ke Python 3.
```
Using Python version 3.6.8
```

Metode permanen

Login ke kluster melalui SSH. Untuk informasi lebih lanjut, lihat Login ke kluster.
Ubah file konfigurasi.
1. Jalankan perintah berikut untuk membuka file profile:
```
vi /etc/profile
```
2. Tekan i untuk memasuki mode insert.
3. Di akhir file profile, tambahkan baris berikut:
```
export PYSPARK_PYTHON=/usr/bin/python3
```
4. Tekan Esc untuk keluar dari mode insert, lalu masukkan :wq untuk menyimpan dan menutup file.
Jalankan perintah berikut untuk memuat ulang file konfigurasi agar perubahan diterapkan segera:
```
source /etc/profile
```
Jalankan perintah berikut untuk memeriksa versi Python:
```
pyspark
```
Jika output berisi pesan berikut, versi Python telah diubah ke Python 3.
```
Using Python version 3.6.8
```

Mengapa pekerjaan Spark Streaming berhenti secara tiba-tiba

Jika versi Spark Anda lebih awal dari 1.6, lakukan upgrade.

Versi Spark sebelum 1.6 memiliki memory leak yang dapat menyebabkan kontainer dihentikan.
Pastikan kode Anda dioptimalkan untuk penggunaan memori.

Mengapa pekerjaan yang dihentikan masih ditampilkan sebagai running

Hal ini dapat terjadi jika Anda mengirimkan pekerjaan dalam mode YARN-client, karena E-MapReduce tidak dapat memantau status pekerjaan secara akurat dalam mode ini. Untuk memastikan pelaporan status yang akurat, kirimkan pekerjaan dalam mode YARN-cluster.

`java.lang.ClassNotFoundException` error untuk `spark-submit` dalam mode YARN-cluster

Berikut adalah contoh pesan error:

Process Output>>>  24/09/30 15:41:24 WARN HiveConf: HiveConf of name hive.metastore.type does not exist
Process Output>>>  24/09/30 15:41:24 ERROR Hive: Unable to instantiate a metastore client factory com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory: java.lang.ClassNotFoundException: Class com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory not found
Process Output>>>  java.lang.ClassNotFoundException: Class com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory not found
Process Output>>>      at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2542)
Process Output>>>      at org.apache.hadoop.hive.ql.metadata.Hive.createMetaStoreClient(Hive.java:3711)
Process Output>>>      at org.apache.hadoop.hive.ql.metadata.Hive.getMSC(Hive.java:3794)
Process Output>>>      at org.apache.hadoop.hive.ql.metadata.Hive.getMSC(Hive.java:3774)
Process Output>>>      at org.apache.hadoop.hive.ql.metadata.Hive.getDelegationToken(Hive.java:3924)
Process Output>>>      at org.apache.spark.sql.hive.security.HiveDelegationTokenProvider.$anonfun$obtainDelegationTokens$4(HiveDelegationTokenProvider.scala:104)
Process Output>>>      at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
Process Output>>>      at org.apache.spark.sql.hive.security.HiveDelegationTokenProvider$$anon$1.run(HiveDelegationTokenProvider.scala:139)
Process Output>>>      at java.security.AccessController.doPrivileged(Native Method)
Process Output>>>      at javax.security.auth.Subject.doAs(Subject.java:422)
Process Output>>>      at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1730)
Process Output>>>      at org.apache.spark.sql.hive.security.HiveDelegationTokenProvider.doAsRealUser(HiveDelegationTokenProvider.scala:138)
Process Output>>>      at org.apache.spark.sql.hive.security.HiveDelegationTokenProvider.obtainDelegationTokens(HiveDelegationTokenProvider.scala:102)
Process Output>>>      at org.apache.spark.deploy.security.HadoopDelegationTokenManager.$anonfun$obtainDelegationTokens$2(HadoopDelegationTokenManager.scala:164)
Process Output>>>      at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:293)
Process Output>>>      at scala.collection.Iterator.foreach(Iterator.scala:943)
Process Output>>>      at scala.collection.Iterator.foreach$(Iterator.scala:943)
Process Output>>>      at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
Process Output>>>      at scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:214)
Process Output>>>      at scala.collection.TraversableLike.flatMap(TraversableLike.scala:293)
Process Output>>>      at scala.collection.TraversableLike.flatMap$(TraversableLike.scala:290)

Penyebab: Pada kluster EMR yang diaktifkan Kerberos, classpath driver tidak secara otomatis diisi dengan JAR yang diperlukan saat pekerjaan dijalankan dalam mode YARN-cluster. Hal ini menyebabkan error ClassNotFoundException.

Solusi: Pada kluster EMR yang diaktifkan Kerberos, Anda harus menggunakan parameter --jars saat mengirimkan pekerjaan dengan spark-submit dalam mode YARN-cluster. Selain JAR aplikasi Anda, Anda juga harus menyertakan semua paket JAR dari direktori /opt/apps/METASTORE/metastore-current/hive2.

Penting

Dalam mode YARN-cluster, semua jalur file dalam parameter --jars harus dipisahkan dengan koma. Direktori tidak didukung.

Sebagai contoh, jika JAR aplikasi Anda adalah /opt/apps/SPARK3/spark3-current/examples/jars/spark-examples_2.12-3.5.3-emr.jar, jalankan perintah spark-submit berikut:

spark-submit --deploy-mode cluster --class org.apache.spark.examples.SparkPi --master yarn \
--jars $(ls /opt/apps/METASTORE/metastore-current/hive2/*.jar | tr '\n' ',') \
/opt/apps/SPARK3/spark3-current/examples/jars/spark-examples_2.12-3.5.3-emr.jar