Common issues and solutions when using Spark - E-MapReduce - Alibaba Cloud Dokümantasyon Merkezi

Bu konu, Spark kullanımıyla ilgili sık sorulan soruları yanıtlar.

Spark Core
Spark SQL
- Spark 3'teki küçük dosya birleştirme özelliğini nasıl kullanırım?
- Spark SQL'de veri eğriliğini nasıl yönetirim?
PySpark
- PySpark için Python 3'ü nasıl belirtirim?
Spark Streaming
- Spark Streaming işim bir süre çalıştıktan sonra neden beklenmedik şekilde duruyor?
- Durdurulan bir Spark Streaming işi EMR konsolunda neden hala Çalışıyor olarak gösteriliyor?
spark-submit
- Kerberos etkinleştirilmiş bir EMR kümesinde YARN-cluster modunda spark-submit çalıştırdığımda neden java.lang.ClassNotFoundException hatası alıyorum?

Geçmiş Spark işlerini nerede görüntüleyebilirim?

EMR konsolunda hedef kümenin Access Links and Ports sekmesine gidin ve geçmiş Spark işlerini görüntülemek için Spark UI bağlantısını tıklayın. Bileşen kullanıcı arayüzlerine erişim hakkında daha fazla bilgi için bkz. EMR konsolunda açık kaynak bileşenlerin web kullanıcı arayüzlerine erişme.

Spark işlerini bağımsız (standalone) modda gönderebilir miyim?

Hayır. E-MapReduce yalnızca Spark on YARN ve Spark on Kubernetes üzerinden iş göndermeyi destekler. Bağımsız (standalone) ve Mesos modları desteklenmez.

Spark 2 CLI günlük çıktısını azaltma

Varsayılan olarak EMR DataLake kümesindeki spark-sql ve spark-shell gibi komut satırı arabirimi (CLI) araçları INFO düzeyinde günlük kaydı yapar.

CLI araçlarını çalıştırdığınız düğümde (örneğin master düğümünde) bir log4j.properties yapılandırma dosyası oluşturun. Varsayılan yapılandırma dosyasını kopyalamak için aşağıdaki komutu kullanın:
```
cp /etc/emr/spark-conf/log4j.properties /new/path/to/log4j.properties
```
Yeni yapılandırma dosyasındaki günlük düzeyini değiştirin.
```
log4j.rootCategory=WARN, console
```
Spark hizmetinin spark-defaults.conf dosyasında spark.driver.extraJavaOptions özelliğini güncelleyin. -Dlog4j.configuration=file:/etc/emr/spark-conf/log4j.properties ifadesini -Dlog4j.configuration=file:/new/path/to/log4j.properties ile değiştirin.

Önemli
Yol file: önekiyle başlamalıdır.

Spark 3 küçük dosya birleştirmeyi kullanma

Küçük dosyaları otomatik olarak birleştirmek için spark.sql.adaptive.merge.output.small.files.enabled parametresini true olarak ayarlayın. Birleştirilen dosyalar sıkıştırılır. Birleştirilen dosyalar çok küçükse spark.sql.adaptive.advisoryOutputFileSizeInBytes parametresinin değerini artırın. Varsayılan değer 256 MB'tır.

Spark SQL'de veri eğriliğini yönetme

Spark 2 için aşağıdaki yaklaşımlardan birini kullanın:

Tabloyu okurken null değerler gibi ilgisiz verileri filtreleyin.

Daha küçük tabloyu yayınlayın (broadcast).

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

Eğri anahtara göre eğri verileri ayırın.

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

Eğri anahtar biliniyorsa verileri dağıtın.

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)

Eğri anahtar bilinmiyorsa verileri dağıtın.

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

Spark 3 için EMR konsolunda Spark 3 hizmetinin Configure sekmesine gidin ve spark.sql.adaptive.enabled ile spark.sql.adaptive.skewJoin.enabled parametrelerini true olarak ayarlayın.

PySpark için Python 3 belirtme

Bu bölümde Spark 2 içeren EMR V5.7.0 üzerindeki bir DataLake kümesi örneği kullanılarak PySpark için Python 3'ün nasıl belirtileceği gösterilmektedir.

Python sürümünü değiştirmenin iki yöntemi vardır:

Geçici yöntem

SSH üzerinden kümeye bağlanın. Daha fazla bilgi için bkz. Kümeye bağlanma.
Python sürümünü değiştirmek için aşağıdaki komutu çalıştırın:
```
export PYSPARK_PYTHON=/usr/bin/python3
```
Python sürümünü kontrol etmek için aşağıdaki komutu çalıştırın:
```
pyspark
```
Çıktıda aşağıdaki mesaj yer alıyorsa Python sürümü Python 3 olarak değişmiştir.
```
Using Python version 3.6.8
```

Kalıcı yöntem

SSH üzerinden kümeye bağlanın. Daha fazla bilgi için bkz. Kümeye bağlanma.
Yapılandırma dosyasını değiştirin.
1. profile dosyasını açmak için aşağıdaki komutu çalıştırın:
```
vi /etc/profile
```
2. Ekleme moduna geçmek için i tuşuna basın.
3. profile dosyasının sonuna aşağıdaki satırı ekleyin:
```
export PYSPARK_PYTHON=/usr/bin/python3
```
4. Ekleme modundan çıkmak için Esc tuşuna basın. Ardından dosyayı kaydetmek ve kapatmak için :wq girin.
Değişiklikleri hemen uygulamak üzere yapılandırma dosyasını yeniden yüklemek için aşağıdaki komutu çalıştırın:
```
source /etc/profile
```
Python sürümünü kontrol etmek için aşağıdaki komutu çalıştırın:
```
pyspark
```
Çıktıda aşağıdaki mesaj yer alıyorsa Python sürümü Python 3 olarak değişmiştir.
```
Using Python version 3.6.8
```

Spark Streaming işlerinin beklenmedik şekilde durması

Spark sürümünüz 1,6'dan eskiyse yükseltin.

1,6'dan önceki Spark sürümlerinde konteynerlerin sonlandırılmasına neden olabilecek bir bellek sızıntısı bulunur.
Kodunuzun bellek kullanımı açısından optimize edildiğinden emin olun.

Durdurulan bir işin çalışıyor olarak görünmesi

İşi YARN-client modunda gönderdiyseniz bu durum oluşabilir çünkü E-MapReduce bu moddaki işlerin durumunu doğru şekilde izleyemez. Doğru durum raporlamasını sağlamak için işi bunun yerine YARN-cluster modunda gönderin.

YARN-cluster modunda spark-submit için `java.lang.ClassNotFoundException` hatası

Aşağıda örnek bir hata mesajı verilmiştir:

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)

Neden: Kerberos etkinleştirilmiş bir EMR kümesinde bir iş YARN-cluster modunda çalıştırıldığında sürücü sınıf yoluna (classpath) gerekli JAR'lar otomatik olarak eklenmez. Bu durum ClassNotFoundException hatasına neden olur.

Çözüm: Kerberos etkinleştirilmiş bir EMR kümesinde YARN-cluster modunda spark-submit ile bir iş gönderirken --jars parametresini kullanmanız gerekir. Uygulamanızın JAR'larına ek olarak /opt/apps/METASTORE/metastore-current/hive2 dizinindeki tüm JAR paketlerini de dahil etmelisiniz.

Önemli

YARN-cluster modunda --jars parametresindeki tüm dosya yolları virgülle ayrılmış olmalıdır. Dizinler desteklenmez.

Örneğin uygulama JAR'ınız /opt/apps/SPARK3/spark3-current/examples/jars/spark-examples_2.12-3.5.3-emr.jar ise aşağıdaki spark-submit komutunu çalıştırın:

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