Common issues and solutions when using Spark - E-MapReduce

This topic answers common questions about using Spark.

Spark Core
Spark SQL
- How do I use the small file merging feature in Spark 3?
- How do I handle data skew in Spark SQL?
PySpark
- How can I specify Python 3 for PySpark?
Spark Streaming
- Why does my Spark Streaming job stop unexpectedly after running for a while?
- Why does a stopped Spark Streaming job still show as Running in the EMR console?
spark-submit
- Why do I get a java.lang.ClassNotFoundException error when I run spark-submit in YARN-cluster mode on a Kerberos-enabled EMR cluster?

Where can I view historical Spark jobs?

In the EMR console, go to the Access Links and Ports tab on the target cluster and click the Spark UI link to view historical Spark jobs. For more information about accessing component UIs, see Access the web UIs of open-source components in the EMR console.

Can I submit Spark jobs in standalone mode?

No. E-MapReduce supports submitting jobs only through Spark on YARN and Spark on Kubernetes. The standalone and Mesos modes are not supported.

How to reduce Spark 2 CLI log output

By default, command-line interface (CLI) tools such as spark-sql and spark-shell on an EMR DataLake cluster output INFO-level logs.

On the node where you run the CLI tools (for example, the master node), create a log4j.properties configuration file. To copy the default configuration file, use the following command:
```
cp /etc/emr/spark-conf/log4j.properties /new/path/to/log4j.properties
```
Modify the log level in the new configuration file.
```
log4j.rootCategory=WARN, console
```
In the spark-defaults.conf file for the Spark service, update the spark.driver.extraJavaOptions property. Replace -Dlog4j.configuration=file:/etc/emr/spark-conf/log4j.properties with -Dlog4j.configuration=file:/new/path/to/log4j.properties.

Important
The path must be prefixed with file:.

How to use Spark 3 small file merging

Set the spark.sql.adaptive.merge.output.small.files.enabled parameter to true to automatically merge small files. The merged files are compressed. If the merged files are too small, increase the value of the spark.sql.adaptive.advisoryOutputFileSizeInBytes parameter. The default value is 256 MB.

How to handle data skew in Spark SQL

For Spark 2, use one of the following approaches:

Filter out irrelevant data, such as null values, when reading a table.

Broadcast the smaller table.

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

Separate skewed data based on the skewed key.

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

If the skewed key is known, scatter the data.

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)

If the skewed key is unknown, scatter the data.

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

For Spark 3, go to the Configure tab of the Spark 3 service in the EMR console, and set the spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled parameters to true.

How to specify Python 3 for PySpark

This section demonstrates how to specify Python 3 for PySpark, using a DataLake cluster on EMR V5.7.0 with Spark 2 as an example.

There are two methods to change the Python version:

Temporary method

Log on to the cluster over SSH. For more information, see Log on to a cluster.
Run the following command to change the Python version:
```
export PYSPARK_PYTHON=/usr/bin/python3
```
Run the following command to check the Python version:
```
pyspark
```
If the output contains the following message, the Python version has been changed to Python 3.
```
Using Python version 3.6.8
```

Permanent method

Log on to the cluster over SSH. For more information, see Log on to a cluster.
Modify the configuration file.
1. Run the following command to open the profile file:
```
vi /etc/profile
```
2. Press i to enter insert mode.
3. At the end of the profile file, add the following line:
```
export PYSPARK_PYTHON=/usr/bin/python3
```
4. Press Esc to exit insert mode. Then, enter :wq to save and close the file.
Run the following command to reload the configuration file to apply the changes immediately:
```
source /etc/profile
```
Run the following command to check the Python version:
```
pyspark
```
If the output contains the following message, the Python version has been changed to Python 3.
```
Using Python version 3.6.8
```

Why Spark Streaming jobs stop unexpectedly

If your Spark version is earlier than 1.6, upgrade it.

Spark versions earlier than 1.6 have a memory leak that can cause containers to be terminated.
Ensure your code is optimized for memory usage.

Why a stopped job appears as running

This can happen if you submitted the job in YARN-client mode, as E-MapReduce cannot accurately monitor the status of jobs in this mode. To ensure correct status reporting, submit the job in YARN-cluster mode instead.

`java.lang.ClassNotFoundException` error for `spark-submit` in YARN-cluster mode

The following is a sample error message:

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)

Cause: On a Kerberos-enabled EMR cluster, the driver classpath is not automatically populated with required JARs when a job is run in YARN-cluster mode. This causes the ClassNotFoundException error.

Solution: On a Kerberos-enabled EMR cluster, you must use the --jars parameter when submitting a job with spark-submit in YARN-cluster mode. In addition to your application's JARs, you must also include all JAR packages from the /opt/apps/METASTORE/metastore-current/hive2 directory.

Important

In YARN-cluster mode, all file paths in the --jars parameter must be comma-separated. Directories are not supported.

For example, if your application JAR is /opt/apps/SPARK3/spark3-current/examples/jars/spark-examples_2.12-3.5.3-emr.jar, run the following spark-submit command:

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