Spark ERROR in cluster: ModuleNotFoundError: No module named 'cst_utils'
Spark ERROR in cluster: ModuleNotFoundError: No module named 'cst_utils'
我有一个带有 python 的 Spark 程序。程序结构是这样的：
cst_utils.py
bn_utils.py
ep_utils.py
main.py
每个 cst_utils.py,bn_utils.py,ep_utils.py 都有一个名为 Spark_Func(sc) 的函数。在 main 中，我创建了一个 Spark 上下文，sc，并将其发送到每个 Spark_Func，如下所示：
import cst_utils as cu
import bn_utils as bu
import ep_utils as eu
spark_conf = SparkConf().setAppName('app_name') \
.setMaster("spark://x.x.x.x:7077") \
.set('spark.executor.memory', "8g") \
.set('spark.executor.cores', 4) \
.set('spark.task.cpus', 2)
sc = SparkContext(conf=spark_conf)
cu.spark_func(sc)
bu.spark_func(sc)
eu.spark_func(sc)
我用两个Slaves和一个Master配置Spark集群，它们都有Ubuntu 20.04 OS。我在 spark-env.sh 中设置了主 IP，并使 SSH 无密码，主节点无需身份验证即可访问每个从节点。我在每个节点中 运行 这些命令：
主节点：
./start-master.sh
奴隶：
./start-worker.sh spark://x.x.x.x:7077
集群已创建，因为我可以在浏览器中使用此命令看到 SPARK UI：
http://x.x.x.x:8080
但是当我想 运行 使用这个命令的程序时：
/opt/spark/bin/spark-submit --master spark://x.x.x.x:7077 main.py
我收到此错误：
22/02/16 16:39:20 INFO SparkContext: Starting job: count at /home/hs/Desktop/etl/cst_utils.py:442
22/02/16 16:39:20 INFO DAGScheduler: Registering RDD 2 (reduceByKey at /home/hs/Desktop/etl/cst_utils.py:434) as input to shuffle 0
22/02/16 16:39:20 INFO DAGScheduler: Got job 0 (count at /home/hs/Desktop/etl/cst_utils.py:442) with 1 output partitions
22/02/16 16:39:20 INFO DAGScheduler: Final stage: ResultStage 1 (count at /home/hs/Desktop/etl/cst_utils.py:442)
22/02/16 16:39:20 INFO DAGScheduler: Parents of final stage: List(ShuffleMapStage 0)
22/02/16 16:39:20 INFO DAGScheduler: Missing parents: List(ShuffleMapStage 0)
22/02/16 16:39:20 INFO DAGScheduler: Submitting ShuffleMapStage 0 (PairwiseRDD[2] at reduceByKey at /home/hs/Desktop/etl/cst_utils.py:434), which has no missing parents
22/02/16 16:39:20 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 9.4 KiB, free 366.3 MiB)
22/02/16 16:39:20 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 5.9 KiB, free 366.3 MiB)
22/02/16 16:39:20 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on x.x.x.x:43875 (size: 5.9 KiB, free: 366.3 MiB)
22/02/16 16:39:20 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1388
22/02/16 16:39:20 INFO DAGScheduler: Submitting 1 missing tasks from ShuffleMapStage 0 (PairwiseRDD[2] at reduceByKey at /home/hs/Desktop/etl/cst_utils.py:434) (first 15 tasks are for partitions Vector(0))
22/02/16 16:39:20 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks resource profile 0
22/02/16 16:39:21 INFO CoarseGrainedSchedulerBackend$DriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (z.z.z.z:39668) with ID 1, ResourceProfileId 0
22/02/16 16:39:21 INFO CoarseGrainedSchedulerBackend$DriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (y.y.y.y:46330) with ID 0, ResourceProfileId 0
22/02/16 16:39:21 INFO BlockManagerMasterEndpoint: Registering block manager y.y.y.y:34159 with 4.1 GiB RAM, BlockManagerId(0, y.y.y.y, 34159, None)
22/02/16 16:39:21 INFO BlockManagerMasterEndpoint: Registering block manager z.z.z.z:42231 with 4.1 GiB RAM, BlockManagerId(1, z.z.z.z, 42231, None)
22/02/16 16:39:21 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (y.y.y.y, executor 0, partition 0, PROCESS_LOCAL, 4481 bytes) taskResourceAssignments Map()
22/02/16 16:39:21 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on y.y.y.y:34159 (size: 5.9 KiB, free: 4.1 GiB)
22/02/16 16:39:22 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0) (y.y.y.y executor 0): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/opt/spark/python/lib/pyspark.zip/pyspark/worker.py", line 586, in main
func, profiler, deserializer, serializer = read_command(pickleSer, infile)
File "/opt/spark/python/lib/pyspark.zip/pyspark/worker.py", line 69, in read_command
command = serializer._read_with_length(file)
File "/opt/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 160, in _read_with_length
return self.loads(obj)
File "/opt/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 430, in loads
return pickle.loads(obj, encoding=encoding)
ModuleNotFoundError: No module named 'cst_utils'
程序路径与所有节点的路径相同，SPARK路径也相同。
事实上，当我运行本地模式下的程序时，它运行没有任何问题。但是，对于 运行 本地，我在 SPARK CONTEXT 中使用此配置：
spark_conf = SparkConf().setAppName('app_name') \
.setMaster("local[4]") \
.set('spark.executor.memory', "8g") \
.set('spark.executor.cores', 4) \
.set('spark.task.cpus', 1)
sc = SparkContext(conf=spark_conf)
更新 1：
我也使用虚拟环境并在其中安装所有包以在节点之间分发它们。详情：
要在 python 中创建虚拟环境 运行 此命令：
sudo apt install python3.8-venv
创建虚拟环境：
python3 -m venv my_venv
进入环境：
source my_vent/bin/activate
我使用 venv-pack 打包您在项目中安装的所有包。
pip install venv-pack
打包：
venv-pack -o my_venv.tar.gz
另外，正如Spark网站所说，我把项目的所有.py文件都放在一个文件夹里，然后压缩成.zip文件夹。
最后在创建集群后，我运行这个命令：
/opt/spark/bin/spark-submit --master spark://x.x.x.x:7077 --archives my_venv.tar.gz#environment --py-files my_files.zip main.py
但是，它以这个错误结束：
Traceback (most recent call last):
File "/home/spark/Desktop/etl/main.py", line 3, in <module>
import cst_utils as cu
File "/home/spark/Desktop/etl/cst_utils.py", line 5, in <module>
import group_state as gs
File "/home/spark/Desktop/etl/group_state.py", line 1, in <module>
import numpy as np
ModuleNotFoundError: No module named 'numpy'
能否请您指导我 运行Cluster 中的 ning 代码有什么问题？
任何帮助将不胜感激。
问题已解决。
首先，我使用以下命令在每个节点中安装了所有软件包：
python3 -m pip install PACKAGE
然后，当我运行程序时，我必须把程序中用到的所有PY文件写在--py-files前面像这样：
/opt/spark/bin/spark-submit --master spark://x.x.x.x:7077 --files sparkConfig.json --py-files cst_utils.py,grouping.py,group_state.py,g_utils.py,csts.py,oracle_connection.py,config.py,brn_utils.py,emp_utils.py main.py
那么导入文件就没有任何错误了
我有一个带有 python 的 Spark 程序。程序结构是这样的：
cst_utils.py
bn_utils.py
ep_utils.py
main.py
每个 cst_utils.py,bn_utils.py,ep_utils.py 都有一个名为 Spark_Func(sc) 的函数。在 main 中，我创建了一个 Spark 上下文，sc，并将其发送到每个 Spark_Func，如下所示：
import cst_utils as cu
import bn_utils as bu
import ep_utils as eu
spark_conf = SparkConf().setAppName('app_name') \
.setMaster("spark://x.x.x.x:7077") \
.set('spark.executor.memory', "8g") \
.set('spark.executor.cores', 4) \
.set('spark.task.cpus', 2)
sc = SparkContext(conf=spark_conf)
cu.spark_func(sc)
bu.spark_func(sc)
eu.spark_func(sc)
我用两个Slaves和一个Master配置Spark集群，它们都有Ubuntu 20.04 OS。我在 spark-env.sh 中设置了主 IP，并使 SSH 无密码，主节点无需身份验证即可访问每个从节点。我在每个节点中 运行 这些命令：
主节点：
./start-master.sh
奴隶：
./start-worker.sh spark://x.x.x.x:7077
集群已创建，因为我可以在浏览器中使用此命令看到 SPARK UI：
http://x.x.x.x:8080
但是当我想 运行 使用这个命令的程序时：
/opt/spark/bin/spark-submit --master spark://x.x.x.x:7077 main.py
我收到此错误：
22/02/16 16:39:20 INFO SparkContext: Starting job: count at /home/hs/Desktop/etl/cst_utils.py:442
22/02/16 16:39:20 INFO DAGScheduler: Registering RDD 2 (reduceByKey at /home/hs/Desktop/etl/cst_utils.py:434) as input to shuffle 0
22/02/16 16:39:20 INFO DAGScheduler: Got job 0 (count at /home/hs/Desktop/etl/cst_utils.py:442) with 1 output partitions
22/02/16 16:39:20 INFO DAGScheduler: Final stage: ResultStage 1 (count at /home/hs/Desktop/etl/cst_utils.py:442)
22/02/16 16:39:20 INFO DAGScheduler: Parents of final stage: List(ShuffleMapStage 0)
22/02/16 16:39:20 INFO DAGScheduler: Missing parents: List(ShuffleMapStage 0)
22/02/16 16:39:20 INFO DAGScheduler: Submitting ShuffleMapStage 0 (PairwiseRDD[2] at reduceByKey at /home/hs/Desktop/etl/cst_utils.py:434), which has no missing parents
22/02/16 16:39:20 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 9.4 KiB, free 366.3 MiB)
22/02/16 16:39:20 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 5.9 KiB, free 366.3 MiB)
22/02/16 16:39:20 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on x.x.x.x:43875 (size: 5.9 KiB, free: 366.3 MiB)
22/02/16 16:39:20 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1388
22/02/16 16:39:20 INFO DAGScheduler: Submitting 1 missing tasks from ShuffleMapStage 0 (PairwiseRDD[2] at reduceByKey at /home/hs/Desktop/etl/cst_utils.py:434) (first 15 tasks are for partitions Vector(0))
22/02/16 16:39:20 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks resource profile 0
22/02/16 16:39:21 INFO CoarseGrainedSchedulerBackend$DriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (z.z.z.z:39668) with ID 1, ResourceProfileId 0
22/02/16 16:39:21 INFO CoarseGrainedSchedulerBackend$DriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (y.y.y.y:46330) with ID 0, ResourceProfileId 0
22/02/16 16:39:21 INFO BlockManagerMasterEndpoint: Registering block manager y.y.y.y:34159 with 4.1 GiB RAM, BlockManagerId(0, y.y.y.y, 34159, None)
22/02/16 16:39:21 INFO BlockManagerMasterEndpoint: Registering block manager z.z.z.z:42231 with 4.1 GiB RAM, BlockManagerId(1, z.z.z.z, 42231, None)
22/02/16 16:39:21 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (y.y.y.y, executor 0, partition 0, PROCESS_LOCAL, 4481 bytes) taskResourceAssignments Map()
22/02/16 16:39:21 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on y.y.y.y:34159 (size: 5.9 KiB, free: 4.1 GiB)
22/02/16 16:39:22 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0) (y.y.y.y executor 0): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/opt/spark/python/lib/pyspark.zip/pyspark/worker.py", line 586, in main
func, profiler, deserializer, serializer = read_command(pickleSer, infile)
File "/opt/spark/python/lib/pyspark.zip/pyspark/worker.py", line 69, in read_command
command = serializer._read_with_length(file)
File "/opt/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 160, in _read_with_length
return self.loads(obj)
File "/opt/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 430, in loads
return pickle.loads(obj, encoding=encoding)
ModuleNotFoundError: No module named 'cst_utils'
程序路径与所有节点的路径相同，SPARK路径也相同。
事实上，当我运行本地模式下的程序时，它运行没有任何问题。但是，对于 运行 本地，我在 SPARK CONTEXT 中使用此配置：
spark_conf = SparkConf().setAppName('app_name') \
.setMaster("local[4]") \
.set('spark.executor.memory', "8g") \
.set('spark.executor.cores', 4) \
.set('spark.task.cpus', 1)
sc = SparkContext(conf=spark_conf)
更新 1：
我也使用虚拟环境并在其中安装所有包以在节点之间分发它们。详情：
要在 python 中创建虚拟环境 运行 此命令：
sudo apt install python3.8-venv创建虚拟环境：
python3 -m venv my_venv进入环境：
source my_vent/bin/activate我使用 venv-pack 打包您在项目中安装的所有包。
pip install venv-pack打包：
venv-pack -o my_venv.tar.gz
另外，正如Spark网站所说，我把项目的所有.py文件都放在一个文件夹里，然后压缩成.zip文件夹。
最后在创建集群后，我运行这个命令：
/opt/spark/bin/spark-submit --master spark://x.x.x.x:7077 --archives my_venv.tar.gz#environment --py-files my_files.zip main.py
但是，它以这个错误结束：
Traceback (most recent call last):
File "/home/spark/Desktop/etl/main.py", line 3, in <module>
import cst_utils as cu
File "/home/spark/Desktop/etl/cst_utils.py", line 5, in <module>
import group_state as gs
File "/home/spark/Desktop/etl/group_state.py", line 1, in <module>
import numpy as np
ModuleNotFoundError: No module named 'numpy'
能否请您指导我 运行Cluster 中的 ning 代码有什么问题？
任何帮助将不胜感激。
问题已解决。
首先，我使用以下命令在每个节点中安装了所有软件包：
python3 -m pip install PACKAGE
然后，当我运行程序时，我必须把程序中用到的所有PY文件写在--py-files前面像这样：
/opt/spark/bin/spark-submit --master spark://x.x.x.x:7077 --files sparkConfig.json --py-files cst_utils.py,grouping.py,group_state.py,g_utils.py,csts.py,oracle_connection.py,config.py,brn_utils.py,emp_utils.py main.py
那么导入文件就没有任何错误了