是什么导致了PySpark中的内存问题？

fiei3ece  于 2024-01-06  发布在  Spark
关注(0)|答案(1)|浏览(174)

我是PySpark的新手，我正在尝试编写一个随机森林回归模型，以根据各种网络指标预测流会话期间发生的缓冲量。该模型将在存储在Hadoop分布式文件系统中的数据集文件上运行。该文件大小为9 MB，用于上下文。
我一直在Oracle VirtualBox上使用两个Ubuntu虚拟机，一个主虚拟机和一个从虚拟机。我遇到了内存问题，我有点困惑为什么会发生这些问题。
代码如下：

  1. #Importing libraries
  2. from pyspark.sql import SparkSession
  3. from pyspark.sql.types import StructType, StructField, IntegerType, FloatType
  4. from pyspark.sql.functions import mean, col
  5. from pyspark.ml.feature import VectorAssembler
  6. from pyspark.ml.regression import RandomForestRegressor
  7. from pyspark.ml import Pipeline
  8. from pyspark.ml.tuning import ParamGridBuilder
  9. from pyspark.ml.tuning import CrossValidator
  10. from pyspark.ml.evaluation import RegressionEvaluator
  11. import pandas as pd
  12. import numpy as np
  14. #Defining fields
  15. Fields = [StructField("NbClients", IntegerType(), True),
  16.     StructField("BottleneckBW", IntegerType(), True),
  17.     StructField("BottleneckDelay", FloatType(), True),
  18.     StructField("ClientResolution", IntegerType(), True),
  19.     StructField("RequestDuration", IntegerType(), True),
  20.     StructField("TCPOutputPacket", FloatType(), True),
  21.     StructField("TCPOutputDelay", FloatType(), True),
  22.     StructField("TCPOutputJitter", FloatType(), True),
  23.     StructField("TCPOutputPloss", IntegerType(), True),
  24.     StructField("TCPInputPacket", FloatType(), True),
  25.     StructField("TCPInputDelay", FloatType(), True),
  26.     StructField("TCPInputJitter", FloatType(), True),
  27.     StructField("TCPInputPloss", IntegerType(), True),
  28.     StructField("TCPInputRetrans", IntegerType(), True),
  29.     StructField("50_InputRateVariation", IntegerType(), True),
  30.     StructField("StdInterATimesReq", FloatType(), True),
  31.     StructField("50_InterATimesReq", FloatType(), True),
  32.     StructField("AvgBufferLevel", FloatType(), True)
  33. ]
  35. #Defining schema
  36. Schema = StructType(fields=Fields)
  38. #Starting Spark session
  39. Spark = SparkSession.builder.master("spark://IPADDRESS:7077").getOrCreate()
  41. #Reading CSV from file system
  42. SparkDataFrame = Spark.read.csv("hdfs://IPADDRESS:9000/Dataset.csv", 
  43. sep=",", header=True, schema=Schema)
  45. #Assembling all the predictive features into one features column that is used by the random forest regressor
  46. Assembler = VectorAssembler(inputCols=[
  47. "NbClients",
  48. "BottleneckBW",
  49. "BottleneckDelay",
  50. "ClientResolution",
  51. "RequestDuration",
  52. "TCPOutputPacket",
  53. "TCPOutputDelay",
  54. "TCPOutputJitter",
  55. "TCPOutputPloss",
  56. "TCPInputPacket",
  57. "TCPInputDelay",
  58. "TCPInputJitter",
  59. "TCPInputPloss",
  60. "TCPInputRetrans",
  61. "50_InputRateVariation",
  62. "StdInterATimesReq",
  63. "50_InterATimesReq"], outputCol="QoSAndContextFeatures",
  64. handleInvalid="keep")
  66. #Creating random forest regressor, with AvgBufferLevel as the label column and the newly created vector QoSAndContextFeatures as the predictive column
  67. RandomForestRegressionModel = RandomForestRegressor(labelCol="AvgBufferLevel", featuresCol="QoSAndContextFeatures")
  69. #Creating a two-stage prediction pipeline that ensures that the data is always passed through the vector assembler prior to the random forest model being created
  70. PredictionPipeline = Pipeline(stages=[Assembler, RandomForestRegressionModel])
  72. #Creating a paramater grid to test for the most optimal hyperparameters of the random forest model
  73. ParameterGrid = ParamGridBuilder() \
  74.             .addGrid(RandomForestRegressionModel.numTrees, [int(x) for x in np.linspace(start=10, stop=100, num=5)]) \
  75.             .addGrid(RandomForestRegressionModel.maxDepth, [int(x) for x in np.linspace(start=5, stop=50, num=5)]) \
  76.             .build()
  78. #Building a cross validator of the random forest model that encapsulates the prediction pipeline and the parameter grid alongside an evaluator
  79. CrossValidation = CrossValidator(estimator=PredictionPipeline,
  80.                              estimatorParamMaps=ParameterGrid,
  81.                              evaluator=RegressionEvaluator(labelCol="AvgBufferLevel"),
  82.                              numFolds=3)
  84. #Splitting the data into training and testing data
  85. (TrainingData, TestingData) = SparkDataFrame.randomSplit([0.8,0.2])
  87. #Fitting the cross validated model to the training data
  88. CrossValidatedModel = CrossValidation.fit(TrainingData)
  90. #Making predictions on the testing data
  91. Predictions = CrossValidatedModel.transform(TestingData)
  93. #Declaring evaluator of RSquared to evaluate the RSquared of the model
  94. RSquaredEvaluator = RegressionEvaluator(labelCol="AvgBufferLevel", predictionCol="Prediction", metric="r2")
  96. #Declaring evaluator of RMSE to evaluate the RMSE of the model
  97. RMSEEvaluator = RegressionEvaluator(labelCol="AvgBufferLevel", predictionCol="Prediction", metric="rmse")
  99. #Calculating the RSquared of the model
  100. RSquared = RSquaredEvaluator.evaluate(Predictions)
  102. #Calculating the RMSE of the model
  103. RMSE = RMSEEvaluator.evaluate(Predictions)
  105. #Putting the mean observation of AvgBufferLevel from the test dataset into an array
  106. MeanObservationArray = TestingData.select(mean(col("AvgBufferLevel"))).collect()
  108. #Collecting the mean observation
  109. MeanObservation = MeanObservationArray[0]
  111. #Calculating the scatter index, which represents RMSE as a proportion of the mean observation
  112. ScatterIndex = RMSE/MeanObservation
  114. #Printing RSquared
  115. print("R Squared:",RSquared)
  117. #Printing RMSE
  118. print("RMSE:",RMSE)
  120. #Printing scatter index
  121. print("Scatter Index:",ScatterIndex)

字符串
我首先尝试在一个总共有8 GB RAM的系统上运行模型，每个VM分配了2GB的RAM。然而，当我尝试在主VM上运行Spark中的代码时，我在执行的中途收到了来自Linux的消息：

  1. Killed

型
我查看了Spark master网站，看看输出了什么错误，错误是沿着行的东西：

  1. ERROR CoarseGrainedBackEndExecutor: Driver disassociated.

型
当我搜索这个错误时，我确定它与没有足够的RAM有关，所以我将每个VM的RAM增加到3GB并再次尝试代码。然而，当我尝试这个时，由于我的系统没有足够的RAM来促进两个Ubuntu VM的运行，从VM在执行中途被VirtualBox中止。
因此，我试着简化了随机森林的代码，删除了两阶段流水线，并在执行前设置了超参数，而不是搜索它们。我想知道这些是否会增加我代码的RAM强度，所以我希望这可以解决这个问题。简化的代码如下：

  1. #Importing libraries
  2. from pyspark.sql import SparkSession
  3. from pyspark.sql.types import StructType, StructField, IntegerType, FloatType
  4. from pyspark.sql.functions import mean, col
  5. from pyspark.ml.feature import VectorAssembler
  6. from pyspark.ml.regression import RandomForestRegressor
  7. from pyspark.ml import Pipeline
  8. from pyspark.ml.tuning import ParamGridBuilder
  9. from pyspark.ml.tuning import CrossValidator
  10. from pyspark.ml.evaluation import RegressionEvaluator
  11. import pandas as pd
  12. import numpy as np
  14. #Defining fields
  15. Fields = [StructField("NbClients", IntegerType(), True),
  16.     StructField("BottleneckBW", IntegerType(), True),
  17.     StructField("BottleneckDelay", FloatType(), True),
  18.     StructField("ClientResolution", IntegerType(), True),
  19.     StructField("RequestDuration", IntegerType(), True),
  20.     StructField("TCPOutputPacket", FloatType(), True),
  21.     StructField("TCPOutputDelay", FloatType(), True),
  22.     StructField("TCPOutputJitter", FloatType(), True),
  23.     StructField("TCPOutputPloss", IntegerType(), True),
  24.     StructField("TCPInputPacket", FloatType(), True),
  25.     StructField("TCPInputDelay", FloatType(), True),
  26.     StructField("TCPInputJitter", FloatType(), True),
  27.     StructField("TCPInputPloss", IntegerType(), True),
  28.     StructField("TCPInputRetrans", IntegerType(), True),
  29.     StructField("50_InputRateVariation", IntegerType(), True),
  30.     StructField("StdInterATimesReq", FloatType(), True),
  31.     StructField("50_InterATimesReq", FloatType(), True),
  32.     StructField("AvgBufferLevel", FloatType(), True)
  33. ]
  35. #Defining schema
  36. Schema = StructType(fields=Fields)
  38. #Starting Spark session
  39. Spark = SparkSession.builder.master("spark://192.168.56.112:7077").getOrCreate()
  41. #Reading CSV from file system
  42. SparkDataFrame = Spark.read.csv("hdfs://192.168.56.112:9000/CleanedAssignmentDataset.csv", 
  43. sep=",", header=True, schema=Schema)
  45. #Assembling all the predictive features into one features column that is used by the random forest regressor
  46. Assembler = VectorAssembler(inputCols=[
  47. "NbClients",
  48. "BottleneckBW",
  49. "BottleneckDelay",
  50. "ClientResolution",
  51. "RequestDuration",
  52. "TCPOutputPacket",
  53. "TCPOutputDelay",
  54. "TCPOutputJitter",
  55. "TCPOutputPloss",
  56. "TCPInputPacket",
  57. "TCPInputDelay",
  58. "TCPInputJitter",
  59. "TCPInputPloss",
  60. "TCPInputRetrans",
  61. "50_InputRateVariation",
  62. "StdInterATimesReq",
  63. "50_InterATimesReq"], outputCol="QoSAndContextFeatures",
  64. handleInvalid="keep")
  66. #Transforming DataFrame into vector with combined features column
  67. AssembledData = Assembler.transform(SparkDataFrame)
  69. #Creating random forest regressor, with AvgBufferLevel as the label column and the newly created vector QoSAndContextFeatures as the predictive column
  70. RandomForestRegressionModel = RandomForestRegressor(labelCol="AvgBufferLevel", featuresCol="QoSAndContextFeatures", numTrees=100, maxDepth=30)
  72. #Splitting the data into training and testing data
  73. (TrainingData, TestingData) = AssembledData.randomSplit([0.8,0.2])
  75. #Fitting the random forest to the training data
  76. RandomForestFitted = RandomForestRegressionModel.fit(TrainingData)
  78. #Making predictions on the testing data
  79. Predictions = RandomForestFitted.transform(TestingData)
  81. #Declaring evaluator of RSquared to evaluate the RSquared of the model
  82. RSquaredEvaluator = RegressionEvaluator(labelCol="AvgBufferLevel", predictionCol="Prediction", metric="r2")
  84. #Declaring evaluator of RMSE to evaluate the RMSE of the model
  85. RMSEEvaluator = RegressionEvaluator(labelCol="AvgBufferLevel", predictionCol="Prediction", metric="rmse")
  87. #Calculating the RSquared of the model
  88. RSquared = RSquaredEvaluator.evaluate(Predictions)
  90. #Calculating the RMSE of the model
  91. RMSE = RMSEEvaluator.evaluate(Predictions)
  93. #Putting the mean observation of AvgBufferLevel from the test dataset into an array
  94. MeanObservationArray = TestingData.select(mean(col("AvgBufferLevel"))).collect()
  96. #Collecting the mean observation
  97. MeanObservation = MeanObservationArray[0]
  99. #Calculating the scatter index, which represents RMSE as a proportion of the mean observation
  100. ScatterIndex = RMSE/MeanObservation
  102. #Printing RSquared
  103. print("R Squared:",RSquared)
  105. #Printing RMSE
  106. print("RMSE:",RMSE)
  108. #Printing scatter index
  109. print("Scatter Index:",ScatterIndex)

型
然而，这段代码仍然返回相同的问题;当我分配2GB时，它仍然被Linux杀死，当我分配3GB时，从VM仍然在执行中途死亡。
因此，我决定在一个有16 GB RAM的系统上尝试代码。我重新配置了这个系统上的VM，每个VM有4GB的RAM，但是当我试图运行简化的代码时，Linux在执行中途返回“Killed”。
因此，我决定尝试将每个VM上的RAM增加到5GB并运行简化的代码。然而，我收到了以下错误：

  1. java.lang.OutOfMemoryError: Java heap space

型
我做了一些研究，我检查了spark-default.conf文件。我的Spark内存默认设置为5GB。
我必须承认，我完全被导致这个错误的原因难住了。我的数据集CSV文件只有9 MB大小。此外，我已经设法用常规Python编写了一个随机森林回归器，使用Scikit Learn库，即使在低功耗的8 GB系统上也能运行良好，即使有参数网格搜索和所有。它花了一两分钟运行，但它运行良好。
有没有人知道是什么导致了我的记忆问题，因为我承认我非常卡住了？
任何建议都非常感谢。

pyspark

1条答案

按热度按时间
c3frrgcw

c3frrgcw1#

我想我现在已经找到了这个问题的根源，并让我的代码工作。
经过进一步调查，我认为问题是由于我对随机森林回归器的超参数有点过于雄心勃勃，当我尝试使用较低的超参数时，代码运行得很好，“杀戮”停止了。
非常感谢大家的帮助！

赞(0）分享  回复(0）举报  2024-01-06
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com