title: Pyspark - Py4JJavaError: An error occurred while calling o102.collectToPython
body: <p>I am working with <a href="https://catalog.data.gov/dataset/u-s-chronic-disease-indicators-cdi" rel="nofollow noreferrer">this</a> data locally with jupyter lab. My spark crashes when I try to do a ranking. I increased my java memory using <a href="http://www.messiahpsychoanalyst.org/wikihow/index.php/How_to_Increase_Java_Memory_in_Windows" rel="nofollow noreferrer">this tutorial</a> too. I don't understand what is going on, I suppose that it's a problem with the drive memory, but why? This is something that I would do easily using pandas. I checked my memory and it doesn't go more that 64% of what I have.</p>
  
  <pre><code>### Spark ###
  # To find out where the pyspark
  import findspark
  findspark.init()
  
  # Creating Spark Context
  from pyspark import SparkContext
  from pyspark.sql import functions as F
  from pyspark.sql import SQLContext
  from pyspark.sql import Window
  from pyspark.sql.types import StructType, StringType, FloatType
  
  ### Working with data ###
  import numpy as np
  import pandas as pd
  
  ### Visualization ###
  import altair as alt
  
  ### Utils ###
  import datetime
  from urllib.request import urlopen
  
  # File format #
  import csv
  import json
  import xml.etree.ElementTree as ET
  
  # Geolocation #
  import geopy
  
  sc = SparkContext("local", "Testing")
  
  spark = SQLContext(sc)
  
  # Get city name
  def get_place(lat, lon):
      geo_key = '' # get key here https://developer.mapquest.com/plan_purchase/steps/business_edition/business_edition_free/register
      geolocator = geopy.geocoders.OpenMapQuest(api_key=geo_key)
      location = geolocator.geocode("%s, %s" % (lat, lon))
      return location.address.split(',')[-3][1:] # Getting city and removing space
  
  # Get lat
  def get_lat(str_in):
      return (str_in[1:-1].split(',')[0])
  
  def get_lon(str_in):
      return (str_in[1:-1].split(',')[1])
  
  # parallelize function to be used with spark
  pl_get_place = F.udf(get_place, StringType())
  pl_get_lat = F.udf(get_lat, StringType())
  pl_get_lon = F.udf(get_lon, StringType())
  
  df = spark.read.csv("data/U.S._Chronic_Disease_Indicators__CDI_.csv", sep=',', header=True)
  
  df_location = df.filter("GeoLocation IS NOT NULL")
  
  # creating latitude and longitude
  df_location = (
      df_location
      .withColumn('lat', pl_get_lat(F.col('GeoLocation')))
      .withColumn('lon', pl_get_lon(F.col('GeoLocation')))
  )
  
  df_location = (
      df_location
      .withColumn("city", pl_get_place(F.col('lat'), F.col('lon')))
  )
  
  top_10_window = Window.partitionBy('Topic').orderBy(F.col('DataValue').desc())
  
  top_10_cities = (
      df_location.select('Topic', 'city', 'DataValue', F.rank().over(top_10_window).alias('rank')) 
      .filter(F.col('rank') &lt;= 10)
  )
  
  to_n_ranks_cities = (
      top_10_cities
      .groupBy('city')
      .agg(F.countDistinct('Topic').alias('num_times_top'))
  ).toPandas()
  </code></pre>
  
  <p>Error:</p>
  
  <pre><code>java.net.SocketException: Connection reset by peer: socket write error
          at java.net.SocketOutputStream.socketWrite0(Native Method)
          at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111)
          at java.net.SocketOutputStream.write(SocketOutputStream.java:155)
          at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
          at java.io.BufferedOutputStream.write(BufferedOutputStream.java:126)
          at java.io.DataOutputStream.write(DataOutputStream.java:107)
          at java.io.FilterOutputStream.write(FilterOutputStream.java:97)
          at org.apache.spark.api.python.PythonRDD$.org$apache$spark$api$python$PythonRDD$$write$1(PythonRDD.scala:211)
          at org.apache.spark.api.python.PythonRDD$$anonfun$writeIteratorToStream$1.apply(PythonRDD.scala:223)
          at org.apache.spark.api.python.PythonRDD$$anonfun$writeIteratorToStream$1.apply(PythonRDD.scala:223)
          at scala.collection.Iterator$class.foreach(Iterator.scala:893)
          at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
          at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:223)
          at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$2.writeIteratorToStream(PythonUDFRunner.scala:52)
          at org.apache.spark.api.python.BasePythonRunner$WriterThread$$anonfun$run$1.apply(PythonRunner.scala:249)
          at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1992)
          at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:172)
  2019-03-12 15:24:16 WARN  TaskSetManager:66 - Lost task 0.0 in stage 1.0 (TID 1, localhost, executor driver): java.net.SocketException: Connection reset by peer: socket write error
          at java.net.SocketOutputStream.socketWrite0(Native Method)
          at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111)
          at java.net.SocketOutputStream.write(SocketOutputStream.java:155)
          at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
          at java.io.BufferedOutputStream.write(BufferedOutputStream.java:126)
          at java.io.DataOutputStream.write(DataOutputStream.java:107)
          at java.io.FilterOutputStream.write(FilterOutputStream.java:97)
          at org.apache.spark.api.python.PythonRDD$.org$apache$spark$api$python$PythonRDD$$write$1(PythonRDD.scala:211)
          at org.apache.spark.api.python.PythonRDD$$anonfun$writeIteratorToStream$1.apply(PythonRDD.scala:223)
          at org.apache.spark.api.python.PythonRDD$$anonfun$writeIteratorToStream$1.apply(PythonRDD.scala:223)
          at scala.collection.Iterator$class.foreach(Iterator.scala:893)
          at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
          at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:223)
          at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$2.writeIteratorToStream(PythonUDFRunner.scala:52)
          at org.apache.spark.api.python.BasePythonRunner$WriterThread$$anonfun$run$1.apply(PythonRunner.scala:249)
          at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1992)
          at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:172)
  </code></pre>