Caleb Pollman This article looks at how to speed up a Python web scraping and crawling script with multithreading via the concurrent.futures module. We'll also break down the script itself and show how to test the parsing functionality with pytest.
After completing this article, you will be able to:
- Scrape and crawl websites with Selenium and parse HTML with Beautiful Soup
- Set up pytest to test the scraping and parsing functionalities
- Execute a web scraper concurrently with the
concurrent.futuresmodule - Configure headless mode for ChromeDriver with Selenium
Contents
Project Setup
Clone down the repo if you'd like to follow along. From the command line run the following commands:
$ git clone [email protected]:testdrivenio/concurrent-web-scraping.git $ cd concurrent-web-scraping $ python -m venv env $ source env/bin/activate (env)$ pip install -r requirements.txt The above commands may differ depending on your environment.
Install ChromeDriver globally. (We're using version 96.0.4664.45).
Script Overview
The script makes 20 requests to Wikipedia:Random -- https://en.wikipedia.org/wiki/Special:Random -- for information about each article using Selenium to automate interaction with the site and Beautiful Soup to parse the HTML.
script.py:
import datetime import sys from time import sleep, time from scrapers.scraper import connect_to_base, get_driver, parse_html, write_to_file def run_process(filename, browser): if connect_to_base(browser): sleep(2) html = browser.page_source output_list = parse_html(html) write_to_file(output_list, filename) else: print("Error connecting to Wikipedia") if __name__ == "__main__": # headless mode? headless = False if len(sys.argv) > 1: if sys.argv[1] == "headless": print("Running in headless mode") headless = True # set variables start_time = time() current_attempt = 1 output_timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") output_filename = f"output_{output_timestamp}.csv" # init browser browser = get_driver(headless=headless) # scrape and crawl while current_attempt <= 20: print(f"Scraping Wikipedia #{current_attempt} time(s)...") run_process(output_filename, browser) current_attempt = current_attempt + 1 # exit browser.quit() end_time = time() elapsed_time = end_time - start_time print(f"Elapsed run time: {elapsed_time} seconds") Let's start with the main block. After determining whether Chrome should run in headless mode and defining a few variables, the browser is initialized via get_driver() from scrapers/scraper.py:
if __name__ == "__main__": # headless mode? headless = False if len(sys.argv) > 1: if sys.argv[1] == "headless": print("Running in headless mode") headless = True # set variables start_time = time() current_attempt = 1 output_timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") output_filename = f"output_{output_timestamp}.csv" ######## # here # ######## # init browser browser = get_driver(headless=headless) # scrape and crawl while current_attempt <= 20: print(f"Scraping Wikipedia #{current_attempt} time(s)...") run_process(output_filename, browser) current_attempt = current_attempt + 1 # exit browser.quit() end_time = time() elapsed_time = end_time - start_time print(f"Elapsed run time: {elapsed_time} seconds") A while loop is then configured to control the flow of the overall scraper.
if __name__ == "__main__": # headless mode? headless = False if len(sys.argv) > 1: if sys.argv[1] == "headless": print("Running in headless mode") headless = True # set variables start_time = time() current_attempt = 1 output_timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") output_filename = f"output_{output_timestamp}.csv" # init browser browser = get_driver(headless=headless) ######## # here # ######## # scrape and crawl while current_attempt <= 20: print(f"Scraping Wikipedia #{current_attempt} time(s)...") run_process(output_filename, browser) current_attempt = current_attempt + 1 # exit browser.quit() end_time = time() elapsed_time = end_time - start_time print(f"Elapsed run time: {elapsed_time} seconds") Within the loop, run_process() is called, which manages the WebDriver connection and scraping functions.
def run_process(filename, browser): if connect_to_base(browser): sleep(2) html = browser.page_source output_list = parse_html(html) write_to_file(output_list, filename) else: print("Error connecting to Wikipedia") In run_process(), the browser instance passed to connect_to_base().
def run_process(filename, browser): ######## # here # ######## if connect_to_base(browser): sleep(2) html = browser.page_source output_list = parse_html(html) write_to_file(output_list, filename) else: print("Error connecting to wikipedia") This function attempts to connect to wikipedia and then uses Selenium's explicit wait functionality to ensure the element with id='content' has loaded before continuing.
def connect_to_base(browser): base_url = "https://en.wikipedia.org/wiki/Special:Random" connection_attempts = 0 while connection_attempts < 3: try: browser.get(base_url) # wait for table element with id = 'content' to load # before returning True WebDriverWait(browser, 5).until( EC.presence_of_element_located((By.ID, "content")) ) return True except Exception as e: print(e) connection_attempts += 1 print(f"Error connecting to {base_url}.") print(f"Attempt #{connection_attempts}.") return False Review the Selenium docs for more information on explicit wait.
To emulate a human user, sleep(2) is called after the browser has connected to Wikipedia.
def run_process(filename, browser): if connect_to_base(browser): ######## # here # ######## sleep(2) html = browser.page_source output_list = parse_html(html) write_to_file(output_list, filename) else: print("Error connecting to Wikipedia") Once the page has loaded and sleep(2) has executed, the browser grabs the HTML source, which is then passed to parse_html().
def run_process(filename, browser): if connect_to_base(browser): sleep(2) ######## # here # ######## html = browser.page_source ######## # here # ######## output_list = parse_html(html) write_to_file(output_list, filename) else: print("Error connecting to Wikipedia") parse_html() uses Beautiful Soup to parse the HTML, generating a list of dicts with the appropriate data.
def parse_html(html): # create soup object soup = BeautifulSoup(html, "html.parser") output_list = [] # parse soup object to get wikipedia article url, title, and last modified date article_url = soup.find("link", {"rel": "canonical"})["href"] article_title = soup.find("h1", {"id": "firstHeading"}).text article_last_modified = soup.find("li", {"id": "footer-info-lastmod"}).text article_info = { "url": article_url, "title": article_title, "last_modified": article_last_modified, } output_list.append(article_info) return output_list This function also passes the article URL to get_load_time(), which loads the URL and records the subsequent load time.
def get_load_time(article_url): try: # set headers headers = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.89 Safari/537.36" } # make get request to article_url response = requests.get( article_url, headers=headers, stream=True, timeout=3.000 ) # get page load time load_time = response.elapsed.total_seconds() except Exception as e: print(e) load_time = "Loading Error" return load_time The output is added to a CSV file.
def run_process(filename, browser): if connect_to_base(browser): sleep(2) html = browser.page_source output_list = parse_html(html) ######## # here # ######## write_to_file(output_list, filename) else: print("Error connecting to Wikipedia") write_to_file():
def write_to_file(output_list, filename): for row in output_list: with open(Path(BASE_DIR).joinpath(filename), "a") as csvfile: fieldnames = ["url", "title", "last_modified"] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writerow(row) Finally, back in the while loop, the current_attempt is incremented and the process starts over again.
if __name__ == "__main__": # headless mode? headless = False if len(sys.argv) > 1: if sys.argv[1] == "headless": print("Running in headless mode") headless = True # set variables start_time = time() current_attempt = 1 output_timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") output_filename = f"output_{output_timestamp}.csv" # init browser browser = get_driver(headless=headless) # scrape and crawl while current_attempt <= 20: print(f"Scraping Wikipedia #{current_attempt} time(s)...") run_process(output_filename, browser) ######## # here # ######## current_attempt = current_attempt + 1 # exit browser.quit() end_time = time() elapsed_time = end_time - start_time print(f"Elapsed run time: {elapsed_time} seconds") Want to test this out? Grab the full script here.
It took about 57 seconds to run:
(env)$ python script.py Scraping Wikipedia #1 time(s)... Scraping Wikipedia #2 time(s)... Scraping Wikipedia #3 time(s)... Scraping Wikipedia #4 time(s)... Scraping Wikipedia #5 time(s)... Scraping Wikipedia #6 time(s)... Scraping Wikipedia #7 time(s)... Scraping Wikipedia #8 time(s)... Scraping Wikipedia #9 time(s)... Scraping Wikipedia #10 time(s)... Scraping Wikipedia #11 time(s)... Scraping Wikipedia #12 time(s)... Scraping Wikipedia #13 time(s)... Scraping Wikipedia #14 time(s)... Scraping Wikipedia #15 time(s)... Scraping Wikipedia #16 time(s)... Scraping Wikipedia #17 time(s)... Scraping Wikipedia #18 time(s)... Scraping Wikipedia #19 time(s)... Scraping Wikipedia #20 time(s)... Elapsed run time: 57.36561393737793 seconds Got it? Great! Let's add some basic testing.
Testing
To test the parsing functionality without initiating the browser and, thus, making repeated GET requests to Wikipedia, you can download the page's HTML (test/test.html) and parse it locally. This can help avoid scenarios where you may get your IP blocked for making too many requests too quickly while writing and testing your parsing functions, as well as saving you time by not needing to fire up a browser every time you run the script.
test/test_scraper.py:
from pathlib import Path import pytest from scrapers import scraper BASE_DIR = Path(__file__).resolve(strict=True).parent @pytest.fixture(scope="module") def html_output(): with open(Path(BASE_DIR).joinpath("test.html"), encoding="utf-8") as f: html = f.read() yield scraper.parse_html(html) def test_output_is_not_none(html_output): assert html_output def test_output_is_a_list(html_output): assert isinstance(html_output, list) def test_output_is_a_list_of_dicts(html_output): assert all(isinstance(elem, dict) for elem in html_output) Ensure all is well:
(env)$ python -m pytest test/test_scraper.py ================================ test session starts ================================= platform darwin -- Python 3.10.0, pytest-6.2.5, py-1.11.0, pluggy-1.0.0 rootdir: /Users/michael/repos/testdriven/async-web-scraping collected 3 items test/test_scraper.py ... [100%] ================================= 3 passed in 0.19 ================================== Want to mock get_load_time() to bypass the GET request?
test/test_scraper_mock.py:
from pathlib import Path import pytest from scrapers import scraper BASE_DIR = Path(__file__).resolve(strict=True).parent @pytest.fixture(scope="function") def html_output(monkeypatch): def mock_get_load_time(url): return "mocked!" monkeypatch.setattr(scraper, "get_load_time", mock_get_load_time) with open(Path(BASE_DIR).joinpath("test.html"), encoding="utf-8") as f: html = f.read() yield scraper.parse_html(html) def test_output_is_not_none(html_output): assert html_output def test_output_is_a_list(html_output): assert isinstance(html_output, list) def test_output_is_a_list_of_dicts(html_output): assert all(isinstance(elem, dict) for elem in html_output) Test:
(env)$ python -m pytest test/test_scraper_mock.py ================================ test session starts ================================= platform darwin -- Python 3.10.0, pytest-6.2.5, py-1.11.0, pluggy-1.0.0 rootdir: /Users/michael/repos/testdriven/async-web-scraping collected 3 items test/test_scraper.py ... [100%] ================================= 3 passed in 0.27s ================================= Configure Multithreading
Now comes the fun part! By making just a few changes to the script, we can speed things up:
import datetime import sys from concurrent.futures import ThreadPoolExecutor, wait from time import sleep, time from scrapers.scraper import connect_to_base, get_driver, parse_html, write_to_file def run_process(filename, headless): # init browser browser = get_driver(headless) if connect_to_base(browser): sleep(2) html = browser.page_source output_list = parse_html(html) write_to_file(output_list, filename) # exit browser.quit() else: print("Error connecting to Wikipedia") browser.quit() if __name__ == "__main__": # headless mode? headless = False if len(sys.argv) > 1: if sys.argv[1] == "headless": print("Running in headless mode") headless = True # set variables start_time = time() output_timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") output_filename = f"output_{output_timestamp}.csv" futures = [] # scrape and crawl with ThreadPoolExecutor() as executor: for number in range(1, 21): futures.append( executor.submit(run_process, output_filename, headless) ) wait(futures) end_time = time() elapsed_time = end_time - start_time print(f"Elapsed run time: {elapsed_time} seconds") With the concurrent.futures library, ThreadPoolExecutor is used to spawn a pool of threads for executing the run_process functions asynchronously. The submit method takes the function along with the parameters for that function and returns a future object. wait is then used to block execution until all tasks are complete.
It's worth noting that you can easily switch to multiprocessing via ProcessPoolExecutor since both ProcessPoolExecutor and ThreadPoolExecutor implement the same interface:
# scrape and crawl with ProcessPoolExecutor() as executor: for number in range(1, 21): futures.append( executor.submit(run_process, output_filename, headless) ) Why multithreading instead of multiprocessing?
Web scraping is I/O bound since the retrieving of the HTML (I/O) is slower than parsing it (CPU). For more on this along with the difference between parallelism (multiprocessing) and concurrency (multithreading), review the Speeding Up Python with Concurrency, Parallelism, and asyncio article.
Run:
(env)$ python script_concurrent.py Elapsed run time: 11.831077098846436 seconds Check out the completed script here.
To speed things up even further we can run Chrome in headless mode by passing in the headless command line argument:
(env)$ python script_concurrent.py headless Running in headless mode Elapsed run time: 6.222846269607544 seconds Conclusion
With a small amount of variation from the original code, we were able to execute the web scraper concurrently to take the script's run time from around 57 seconds to just over 6 seconds. In this specific scenario that's just about 90% faster, which is a huge improvement.
I hope this helps your scripts. You can find the code in the repo. Cheers!