A rough estimate of the worst case memory requirement for driver is
about 2 * k * runs * numFeatures * numPartitions * 8 bytes. I put 2 at
the beginning because the previous centers are still in memory while
receiving new center updates. -Xiangrui
On Fri, Jun 19, 2015 at 9:02 AM, Rogers Jeffrey
<rogers.john2...@gmail.com> wrote:
> Thanks. Setting the driver memory property worked for K=1000 . But when I
> increased K to1500 I get the following error:
>
> 15/06/19 09:38:44 INFO ContextCleaner: Cleaned accumulator 7
>
> 15/06/19 09:38:44 INFO BlockManagerInfo: Removed broadcast_34_piece0 on
> 172.31.3.51:45157 in memory (size: 1568.0 B, free: 10.4 GB)
>
> 15/06/19 09:38:44 INFO BlockManagerInfo: Removed broadcast_34_piece0 on
> 172.31.9.50:59356 in memory (size: 1568.0 B, free: 73.6 GB)
>
> 15/06/19 09:38:44 INFO BlockManagerInfo: Removed broadcast_34_piece0 on
> 172.31.9.50:60934 in memory (size: 1568.0 B, free: 73.6 GB)
>
> 15/06/19 09:38:44 INFO BlockManagerInfo: Removed broadcast_34_piece0 on
> 172.31.15.51:37825 in memory (size: 1568.0 B, free: 73.6 GB)
>
> 15/06/19 09:38:44 INFO BlockManagerInfo: Removed broadcast_34_piece0 on
> 172.31.15.51:60610 in memory (size: 1568.0 B, free: 73.6 GB)
>
> 15/06/19 09:38:44 INFO ContextCleaner: Cleaned shuffle 5
>
> Exception in thread "Thread-2" java.lang.OutOfMemoryError: Requested array
> size exceeds VM limit
>
> at java.util.Arrays.copyOf(Arrays.java:2367)
>
> at
> java.lang.AbstractStringBuilder.expandCapacity(AbstractStringBuilder.java:130)
>
> at
> java.lang.AbstractStringBuilder.ensureCapacityInternal(AbstractStringBuilder.java:114)
>
> at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:587)
>
> at java.lang.StringBuilder.append(StringBuilder.java:214)
>
> at py4j.Protocol.getOutputCommand(Protocol.java:305)
>
> at py4j.commands.CallCommand.execute(CallCommand.java:82)
>
> at py4j.GatewayConnection.run(GatewayConnection.java:207)
>
> at java.lang.Thread.run(Thread.java:745)
>
> Exception in thread "Thread-300" java.lang.OutOfMemoryError: Requested array
> size exceeds VM limit
>
> at java.util.Arrays.copyOf(Arrays.java:2367)
>
> at
> java.lang.AbstractStringBuilder.expandCapacity(AbstractStringBuilder.java:130)
>
> at
> java.lang.AbstractStringBuilder.ensureCapacityInternal(AbstractStringBuilder.java:114)
>
> at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:587)
>
> at java.lang.StringBuilder.append(StringBuilder.java:214)
>
> at py4j.Protocol.getOutputCommand(Protocol.java:305)
>
> at py4j.commands.CallCommand.execute(CallCommand.java:82)
>
> at py4j.GatewayConnection.run(GatewayConnection.java:207)
>
>
> Is there any method/guideline through which I can understand the memory
> requirement before hand and make appropriate configurations?
>
> Regards,
> Rogers Jeffrey L
>
> On Thu, Jun 18, 2015 at 8:14 PM, Rogers Jeffrey <rogers.john2...@gmail.com>
> wrote:
>>
>> I am submitting the application from a python notebook. I am launching
>> pyspark as follows:
>>
>> SPARK_PUBLIC_DNS=ec2-54-165-202-17.compute-1.amazonaws.com
>> SPARK_WORKER_CORES=8 SPARK_WORKER_MEMORY=15g SPARK_MEM=30g OUR_JAVA_MEM=30g
>> SPARK_DAEMON_JAVA_OPTS="-XX:MaxPermSize=30g -Xms30g -Xmx30g" IPYTHON=1
>> PYSPARK_PYTHON=/usr/bin/python SPARK_PRINT_LAUNCH_COMMAND=1
>> ./spark/bin/pyspark --master
>> spark://54.165.202.17.compute-1.amazonaws.com:7077 --deploy-mode client
>>
>> I guess I should be adding another extra argument --conf
>> "spark.driver.memory=15g" . Is that correct?
>>
>> Regards,
>> Rogers Jeffrey L
>>
>> On Thu, Jun 18, 2015 at 7:50 PM, Xiangrui Meng <men...@gmail.com> wrote:
>>>
>>> With 80,000 features and 1000 clusters, you need 80,000,000 doubles to
>>> store the cluster centers. That is ~600MB. If there are 10 partitions,
>>> you might need 6GB on the driver to collect updates from workers. I
>>> guess the driver died. Did you specify driver memory with
>>> spark-submit? -Xiangrui
>>>
>>> On Thu, Jun 18, 2015 at 12:22 PM, Rogers Jeffrey
>>> <rogers.john2...@gmail.com> wrote:
>>> > Hi All,
>>> >
>>> > I am trying to run KMeans clustering on a large data set with 12,000
>>> > points
>>> > and 80,000 dimensions. I have a spark cluster in Ec2 stand alone mode
>>> > with
>>> > 8 workers running on 2 slaves with 160 GB Ram and 40 VCPU.
>>> >
>>> > My Code is as Follows:
>>> >
>>> > def convert_into_sparse_vector(A):
>>> > non_nan_indices=np.nonzero(~np.isnan(A) )
>>> > non_nan_values=A[non_nan_indices]
>>> > dictionary=dict(zip(non_nan_indices[0],non_nan_values))
>>> > return Vectors.sparse (len(A),dictionary)
>>> >
>>> > X=[convert_into_sparse_vector(A) for A in complete_dataframe.values ]
>>> > sc=SparkContext(appName="parallel_kmeans")
>>> > data=sc.parallelize(X,10)
>>> > model = KMeans.train(data, 1000, initializationMode="k-means||")
>>> >
>>> > where complete_dataframe is a pandas data frame that has my data.
>>> >
>>> > I get the error: Py4JNetworkError: An error occurred while trying to
>>> > connect
>>> > to the Java server.
>>> >
>>> > The error trace is as follows:
>>> >
>>> >> ---------------------------------------- Exception happened during
>>> >> processing of request from ('127.0.0.1', 41360) Traceback (most recent
>>> >> call last): File "/usr/lib64/python2.6/SocketServer.py", line 283,
>>> >> in _handle_request_noblock
>>> >> self.process_request(request, client_address) File
>>> >> "/usr/lib64/python2.6/SocketServer.py", line 309, in process_request
>>> >> self.finish_request(request, client_address) File
>>> >> "/usr/lib64/python2.6/SocketServer.py", line 322, in finish_request
>>> >> self.RequestHandlerClass(request, client_address, self) File
>>> >> "/usr/lib64/python2.6/SocketServer.py", line 617, in __init__
>>> >> self.handle() File "/root/spark/python/pyspark/accumulators.py",
>>> >> line 235, in handle
>>> >> num_updates = read_int(self.rfile) File
>>> >> "/root/spark/python/pyspark/serializers.py", line 544, in read_int
>>> >> raise EOFError EOFError
>>> >> ----------------------------------------
>>> >>
>>> >>
>>> >> ---------------------------------------------------------------------------
>>> >> Py4JNetworkError Traceback (most recent call
>>> >> last) <ipython-input-13-3dd00c2c5e93> in <module>()
>>> >> ----> 1 model = KMeans.train(data, 1000,
>>> >> initializationMode="k-means||")
>>> >>
>>> >> /root/spark/python/pyspark/mllib/clustering.pyc in train(cls, rdd, k,
>>> >> maxIterations, runs, initializationMode, seed, initializationSteps,
>>> >> epsilon)
>>> >> 134 """Train a k-means clustering model."""
>>> >> 135 model = callMLlibFunc("trainKMeansModel",
>>> >> rdd.map(_convert_to_vector), k, maxIterations,
>>> >> --> 136 runs, initializationMode, seed,
>>> >> initializationSteps, epsilon)
>>> >> 137 centers = callJavaFunc(rdd.context,
>>> >> model.clusterCenters)
>>> >> 138 return KMeansModel([c.toArray() for c in centers])
>>> >>
>>> >> /root/spark/python/pyspark/mllib/common.pyc in callMLlibFunc(name,
>>> >> *args)
>>> >> 126 sc = SparkContext._active_spark_context
>>> >> 127 api = getattr(sc._jvm.PythonMLLibAPI(), name)
>>> >> --> 128 return callJavaFunc(sc, api, *args)
>>> >> 129
>>> >> 130
>>> >>
>>> >> /root/spark/python/pyspark/mllib/common.pyc in callJavaFunc(sc, func,
>>> >> *args)
>>> >> 119 """ Call Java Function """
>>> >> 120 args = [_py2java(sc, a) for a in args]
>>> >> --> 121 return _java2py(sc, func(*args))
>>> >> 122
>>> >> 123
>>> >>
>>> >> /root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in
>>> >> __call__(self, *args)
>>> >> 534 END_COMMAND_PART
>>> >> 535
>>> >> --> 536 answer = self.gateway_client.send_command(command)
>>> >> 537 return_value = get_return_value(answer,
>>> >> self.gateway_client,
>>> >> 538 self.target_id, self.name)
>>> >>
>>> >> /root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in
>>> >> send_command(self, command, retry)
>>> >> 367 if retry:
>>> >> 368 #print_exc()
>>> >> --> 369 response = self.send_command(command)
>>> >> 370 else:
>>> >> 371 response = ERROR
>>> >>
>>> >> /root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in
>>> >> send_command(self, command, retry)
>>> >> 360 the Py4J protocol.
>>> >> 361 """
>>> >> --> 362 connection = self._get_connection()
>>> >> 363 try:
>>> >> 364 response = connection.send_command(command)
>>> >>
>>> >> /root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in
>>> >> _get_connection(self)
>>> >> 316 connection = self.deque.pop()
>>> >> 317 except Exception:
>>> >> --> 318 connection = self._create_connection()
>>> >> 319 return connection
>>> >> 320
>>> >>
>>> >> /root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in
>>> >> _create_connection(self)
>>> >> 323 connection = GatewayConnection(self.address,
>>> >> self.port,
>>> >> 324 self.auto_close, self.gateway_property)
>>> >> --> 325 connection.start()
>>> >> 326 return connection
>>> >> 327
>>> >>
>>> >> /root/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in
>>> >> start(self)
>>> >> 430 'server'
>>> >> 431 logger.exception(msg)
>>> >> --> 432 raise Py4JNetworkError(msg)
>>> >> 433
>>> >> 434 def close(self):
>>> >>
>>> >> Py4JNetworkError: An error occurred while trying to connect to the
>>> >> Java server
>>> >
>>> >
>>> > Is there any specific setting that I am missing , that causes this
>>> > error?
>>> >
>>> > Thanks and Regards,
>>> > Rogers Jeffrey L
>>
>>
>
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
For additional commands, e-mail: user-h...@spark.apache.org
