Building a Web Scraping Pipeline with ProxyTee and Apache Airflow

Organizing your workflow when building a Python data scraper can often be a challenge. In this tutorial, we'll explore how to streamline your scraping process using ProxyTee and Apache Airflow. First, we’ll demonstrate how to leverage ProxyTee’s Unlimited Residential Proxies for efficient data extraction. Then, we’ll guide you through setting up recurring jobs with Apache Airflow to automate your scraping tasks.

Overview of ProxyTee and Its Features

ProxyTee is a leading provider of rotating residential proxies, offering an affordable, reliable, and easy-to-use solution for various internet activities. With Unlimited Residential Proxies, you can benefit from:

• Unlimited Bandwidth: No need to worry about data overages, even with high-traffic tasks.
• Global IP Coverage: Access over 20 million IP addresses from 100+ countries, enabling you to target specific regions.
• Multiple Protocol Support: Compatible with both HTTP and SOCKS5 protocols, providing flexibility for various applications.
• User-Friendly Interface: Get started quickly with an intuitive GUI that requires minimal technical expertise.
• Auto Rotation: IP addresses automatically rotate at intervals of 3 to 60 minutes, crucial for avoiding detection during web scraping.
• API Integration: Seamlessly integrate ProxyTee into your applications and automate tasks using the simple API.

Acquiring Data with ProxyTee

For this tutorial, we'll simulate acquiring data from an e-commerce website. Specifically, we’ll focus on extracting key details like the product name, description, and price using ProxyTee's rotating residential proxies. The strength of ProxyTee lies in its ability to provide a large pool of IPs for scraping, allowing seamless access to public web data with a cheap price, which is as low as 50% compared to competitors like Bright Data and Oxylabs.

Using a Scraping API Wrapper

Let's begin by creating a Python class to interact with the ProxyTee API. While ProxyTee offers a direct API, you can use the following general structure as a reference, replacing API calls as necessary:

import requests

JOB_STATUS_DONE = "done"
HTTP_NO_CONTENT = 204

class Client:
    def __init__(self, api_key):
        self.api_key = api_key

    def create_jobs(self, urls):
        payload = {
            "urls": urls,
            "parsing_instructions": {
                "title": {
                    "_fns": [
                        {"_fn": "css_one", "_args": ["h2"]},
                        {"_fn": "element_text"},
                    ],
                },
                "price": {
                    "_fns": [
                        {"_fn": "css_one", "_args": [".price"]},
                        {"_fn": "element_text"},
                    ],
                },
                "availability": {
                    "_fns": [
                        {"_fn": "css_one", "_args": [".availability"]},
                        {"_fn": "element_text"},
                    ],
                },
            },
        }
        
        # Example using a generic POST request, adjust to ProxyTee API
        response = requests.post(
            "https://api.proxytee.com/v1/queries/batch", # Example endpoint
            headers={'Authorization': f'Bearer {self.api_key}'},
            json=payload,
        )
        return response.json()

    def is_status_done(self, job_id):
        # Example using a generic GET request, adjust to ProxyTee API
        response = requests.get(
           f"https://api.proxytee.com/v1/queries/{job_id}", # Example endpoint
            headers={'Authorization': f'Bearer {self.api_key}'},
        )
        job_status_data = response.json()
        return job_status_data["status"] == JOB_STATUS_DONE

    def fetch_content_list(self, job_id):
       # Example using a generic GET request, adjust to ProxyTee API
        job_result_response = requests.get(
            f"https://api.proxytee.com/v1/queries/{job_id}/results", # Example endpoint
            headers={'Authorization': f'Bearer {self.api_key}'},
        )
        if job_result_response.status_code == HTTP_NO_CONTENT:
            return None
        job_results_json = job_result_response.json()
        return job_results_json["results"]

This Client class includes three methods:

• create_jobs: Submits URLs for scraping.
• is_status_done: Checks the status of a submitted URL.
• fetch_content_list: Retrieves scraped content.

Setting Up the Job Queue

To manage queued scraping jobs, you'll need a database. Here’s a basic PostgreSQL setup that stores job details:

create sequence queue_seq;

create table queue (
    id int check (id > 0) primary key default nextval('queue_seq'),
    created_at timestamp(0) not null DEFAULT CURRENT_TIMESTAMP,
    updated_at timestamp(0) not null DEFAULT CURRENT_TIMESTAMP,
    status varchar(255) not null DEFAULT 'pending',
    job_id varchar(255)
);

A Python class, Queue, interacts with this database:

import atexit
import psycopg2.extras

STATUS_PENDING = 'pending'
STATUS_COMPLETE = 'complete'
STATUS_DELETED = 'deleted'

class Queue:
    def __init__(self, connection):
        self.connection = connection
        atexit.register(self.cleanup)

    def setup(self):
      # (implementation as per original blog)
        pass
    
    def push(self, job_id):
        self.__execute_and_commit('insert into queue (job_id) values (%s)', [job_id])

    def pull(self):
      # (implementation as per original blog)
        pass

    def delete(self, job_id):
        self.__change_status(job_id, STATUS_DELETED)

    def complete(self, job_id):
        self.__change_status(job_id, STATUS_COMPLETE)

    def touch(self, job_id):
        self.__execute_and_commit('update queue set updated_at = now() where job_id = %s', [job_id])

    def __change_status(self, job_id, status):
        self.__execute_and_commit('update queue set status = %s where job_id = %s', [status, job_id])

    def __execute_and_commit(self, sql, val):
       # (implementation as per original blog)
        pass

    def cleanup(self):
        self.connection.commit()

The main methods include: setup,push, and pull. See the complete code to get all method implementations. The code includes also methods to interact with database such as delete,complete,touch and database helpers __change_status, __execute_and_commit.

Bootstrapping Your Environment

A bootstrap file initializes the Queue and Client classes, fetching API keys and database credentials from environment variables:

import os
import psycopg2
from messenger import Queue
from oxylabs import Client

DB_HOST = os.getenv('DB_HOST', 'postgres')
DB_USER = os.getenv('DB_USER', 'airflow')
DB_PASS = os.getenv('DB_PASS', 'airflow')
DB_NAME = os.getenv('DB_NAME', 'scraper')

MYPROXY_api_key = os.getenv('MYPROXY_api_key', 'your-proxytee-api-key') # Replace with real api key.

connection = psycopg2.connect(
    host=DB_HOST,
    user=DB_USER,
    password=DB_PASS,
    database=DB_NAME
)

queue = Queue(connection)
client = Client(MYPROXY_api_key)

Creating and Running the Scripts

Once these are set, you'll need scripts for different stages:

• setup.py: Creates the database schema (if it doesn’t exist).
• pusher.py: Submits scraping jobs and pushes them into the queue using ProxyTee API
• puller.py: Checks the status of jobs and pulls scraped data.

Integrating with Apache Airflow

Apache Airflow enables you to automate these scripts. Follow Airflow’s official tutorial to set up your environment. Modify the docker-compose.yaml file to include the src folder as a volume. After, you should get a set up with folder structures similar as:

|-- dags
|-- docker-compose.yaml
|-- .env
|-- logs
|-- plugins
|-- src
|   |-- bootstrap.py
|   |-- messenger.py
|   |-- oxylabs.py # You could use this for custom parsing.
|   |-- puller.py
|   |-- pusher.py
|   |-- setup.py

You can access the Airflow UI at http://localhost:8080 and use the default credentials (airflow:airflow) to access it.

Defining Tasks in Airflow DAGs

Airflow organizes tasks using Directed Acyclic Graphs (DAGs). Define DAGs in Python scripts, such as:

setup.py

from datetime import timedelta
import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(hours=3),
}

with DAG(
        'setup',
        default_args=default_args,
        schedule_interval='@once',
        description='Setup',
        start_date=pendulum.datetime(2022, 5, 1, tz='UTC'), # The year 2022 was used here, please make your adjustment.
        dagrun_timeout=timedelta(minutes=1),
        tags=['scrape', 'database'],
        catchup=False
) as dag:
    setup_task = BashOperator(
        task_id='setup',
        bash_command='python /opt/airflow/src/setup.py',
    )

This creates a DAG that sets up the database schema. Let’s move into a more practical scenario: pushing and pulling scraping jobs. In next section, push-pull.py DAG helps you accomplish this

push-pull.py

from datetime import timedelta
import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'retries': 2,
    'retry_delay': timedelta(hours=3),
}

with DAG(
        'push_pull',
        default_args=default_args,
        schedule_interval='@daily',
        description='Push-Pull workflow',
        start_date=pendulum.datetime(2022, 5, 1, tz='UTC'), # The year 2022 was used here, please make your adjustment.
        dagrun_timeout=timedelta(minutes=1),
        tags=['scrape', 'database'],
        catchup=False
) as dag:
    task_push = BashOperator(
        task_id='push',
        bash_command='python /opt/airflow/src/pusher.py',
    )

    task_pull = BashOperator(
        task_id='pull',
        bash_command='python /opt/airflow/src/puller.py'
    )

    task_push.set_downstream(task_pull)

Utilizing ShortCircuitOperator

The ShortCircuitOperator allows you to skip tasks based on a certain conditions, such as execute only once per day.

push-pull-reworked.py

from datetime import timedelta
import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import ShortCircuitOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'retries': 2,
    'retry_delay': timedelta(hours=3),
}

with DAG(
        'push-pull-reworked',
        default_args=default_args,
        schedule_interval='* * * * *',
        description='Scrape the website',
        start_date=pendulum.datetime(2022, 5, 1, tz='UTC'), # The year 2022 was used here, please make your adjustment.
        dagrun_timeout=timedelta(minutes=1),
        tags=['scrape', 'oxylabs', 'push', 'pull'],
        catchup=False
) as dag:
    def is_midnight(logical_date):
        return logical_date.hour == 0 and logical_date.minute == 0

    trigger_once_per_day = ShortCircuitOperator(
        task_id='once_per_day',
        python_callable=is_midnight,
        provide_context=True,
        dag=dag
    )

    task_push = BashOperator(
        task_id='push',
        bash_command='python /opt/airflow/src/pusher.py',
    )

    trigger_once_per_day.set_downstream(task_push)


    task_pull = BashOperator(
        task_id='pull',
        bash_command='python /opt/airflow/src/puller.py'
    )

Here we can skip tasks based on condition. For this, you can define helper function such as is_midnight.

Combining all in one dag by utilizing ShortCircuitOperator is also feasible and efficient as showcased in file scrape.py, please find content as below:

scrape.py

from datetime import timedelta
import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import ShortCircuitOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'retries': 2,
    'retry_delay': timedelta(hours=3),
}

with DAG(
        'scrape',
        default_args=default_args,
        schedule_interval='* * * * *',
        description='Scrape the website',
        start_date=pendulum.datetime(2022, 5, 1, tz='UTC'), # The year 2022 was used here, please make your adjustment.
        dagrun_timeout=timedelta(minutes=1),
        tags=['scrape', 'oxylabs', 'push', 'pull'],
        catchup=False
) as dag:
    trigger_always = ShortCircuitOperator(
        task_id='always',
        python_callable=lambda prev_start_date_success: prev_start_date_success is not None,
        provide_context=True,
        dag=dag
    )

    trigger_once = ShortCircuitOperator(
        task_id='once',
        python_callable=lambda prev_start_date_success: prev_start_date_success is None,
        provide_context=True,
        dag=dag
    )


    setup_task = BashOperator(
        task_id='setup',
        bash_command='python /opt/airflow/src/setup.py',
    )


    trigger_once.set_downstream(setup_task)

    def is_midnight(logical_date):
        return logical_date.hour == 0 and logical_date.minute == 0

    trigger_once_per_day = ShortCircuitOperator(
        task_id='once_per_day',
        python_callable=is_midnight,
        provide_context=True,
        dag=dag
    )


    task_push = BashOperator(
        task_id='push',
        bash_command='python /opt/airflow/src/pusher.py',
    )

    trigger_once_per_day.set_downstream(task_push)

    task_pull = BashOperator(
        task_id='pull',
        bash_command='python /opt/airflow/src/puller.py'
    )

    trigger_always.set_downstream(task_pull)

    trigger_always.set_downstream(trigger_once_per_day)

This is a flexible design. You could further adjust the logic to trigger your scraping pipeline

Summary

Automating your web scraping workflow with ProxyTee and Apache Airflow makes it simpler, more efficient and cost-effective. ProxyTee’s robust rotating residential proxies—along with auto-rotation and global coverage—provide a reliable base for your scraping tasks. Couple it with the automation and task management features of Apache Airflow, and you have an efficient and adaptable setup for your scraping needs. Start exploring the combination to fully realize their benefits. ProxyTee provides a competitive price, try it for free!