title: How do I download a large list of URLs in parallel in pyspark?
body: <p>I have an RDD containing 10000 urls to be fetched.</p>
  
  <pre><code>list = ['http://SDFKHSKHGKLHSKLJHGSDFKSJH.com',
          'http://google.com',
          'http://twitter.com']
  urls = sc.parallelize(list)
  </code></pre>
  
  <p>I need to check which urls are broken and preferably fetch the results to a corresponding RDD in Python. I tried this:</p>
  
  <pre><code>import asyncio
  import concurrent.futures
  import requests
  
  async def get(url):
  
      with concurrent.futures.ThreadPoolExecutor() as executor:
  
          loop = asyncio.get_event_loop()
          futures = [
              loop.run_in_executor(
                  executor, 
                  requests.get, 
                  i
              )
              for i in url
          ]
          return futures
  
  async def get_response(futures):
      response = await asyncio.gather(futures,return_exceptions=True)
      return(response)
  
  tasks = urls.map(lambda query: get(query)) # Method returns http call response as a Future[String]
  
  results = tasks.map(lambda task: get_response(task) )
  results = results.map(lambda response:'ERR' if isinstance(response, Exception) else 'OK' )
  results.collect()
  </code></pre>
  
  <p>I get the following output which obviously is not right:</p>
  
  <pre><code>['OK', 'OK', 'OK']
  </code></pre>
  
  <p>I also tried this:</p>
  
  <pre><code>import asyncio
  import concurrent.futures
  import requests
  
  async def get():
  
      with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
  
          loop = asyncio.get_event_loop()
          futures = [
              loop.run_in_executor(
                  executor, 
                  requests.get, 
                  i
              )
              for i in urls.toLocalIterator()
          ]
          for response in await asyncio.gather(*futures,return_exceptions=True):
              print('{}: {}'.format(response, 'ERR' if isinstance(response, Exception) else 'OK'))
              pass
  
  
  loop = asyncio.get_event_loop()
  loop.run_until_complete(get())
  </code></pre>
  
  <p>I get the following output: </p>
  
  <pre><code>HTTPConnectionPool(host='SDFKHSKHGKLHSKLJHGSDFKSJH.com', port=80): Max retries exceeded with url: / (Caused by 
  NewConnectionError('&lt;urllib3.connection.HTTPConnection object at 0x12c834210&gt;: Failed to establish a new connection: [Errno 8] nodename nor servname provided, or not known')): ERR
  &lt;Response [200]&gt;: OK
  &lt;Response [200]&gt;: OK
  </code></pre>
  
  <p>Desired output would be something like this:</p>
  
  <pre><code>http://SDFKHSKHGKLHSKLJHGSDFKSJH.com : ERR
  http://google.com : OK
  http://twitter.com : OK
  </code></pre>
  
  <p>But the problem with the second approach is the use of lists to store future objects. I believe that using RDD is better, since number of urls can be in millions or billions and no singe machine can handle it. Also it is not clear to me how to retrieve urls from responses. </p>