--- title: "Spark Structured Streaming with continuous web data ingestion" slug: spark-structured-streaming-with-web-data date: 2026-03-30T07:28:37+00:00 modified: 2026-03-30T08:21:05+00:00 permalink: https://brightdata.com/blog/web-data/spark-structured-streaming-with-web-data type: blog --- [ Blog ](https://brightdata.com/blog "Blog") / [Web Data](https://brightdata.com/blog/web-data) [Web Data](https://brightdata.com/blog/web-data) # Spark Structured Streaming with continuous web data ingestion This tutorial walks you through building a continuous SERP ingestion pipeline with Apache Spark Structured Streaming and Bright Data’s SERP API. 14 min read [ ](https://brightdata.com/blog/authors/arindam-majumder) [Arindam Majumder Technical Writer ](https://brightdata.com/blog/authors/arindam-majumder) ![Spark Structured Streaming with continuous web data ingestion](https://media.brightdata.com/2026/03/Spark-Structured-Streaming-with-continuous-web-data-ingestion.png) In this article, you will learn: - What Apache Spark Structured Streaming is and what it offers. - Why integrating [Bright Data’s SERP API](/products/serp-api) into a Spark Structured Streaming pipeline is a winning strategy. - How to build a PySpark pipeline that continuously ingests live web search data using Bright Data’s SERP API. Let’s jump in! ## What Is Apache Spark Structured Streaming? Apache Spark Structured Streaming is a scalable, fault-tolerant stream processing engine built on top of the Spark SQL engine. Unlike the older Spark Streaming library (which divides data into discrete RDD-based micro-batches using DStreams), Structured Streaming treats a live data stream as an unbounded table that is continuously appended to. You write the same DataFrame and SQL API code you would write for a static batch job, and Spark takes care of running it incrementally as new data arrives. The engine operates in a micro-batch execution model by default. At each trigger interval, Spark reads the latest data from the source, processes it, and writes results to a sink. It tracks progress via checkpointing, so the pipeline can recover from failures and resume exactly where it left off, providing end-to-end, fault-tolerant guarantees. Structured Streaming supports a variety of built-in sources: Kafka topics, Delta tables, cloud object storage via Auto Loader, rate generators (for testing), and more. For sources not natively covered (such as a REST API), you can use the `foreachBatch` extension method, which hands each micro-batch to a Python function where you can express arbitrary ingestion logic. This is the approach we will use here. ### Spark Streaming vs. Spark Structured Streaming: What’s the Difference? If you are familiar with the legacy Spark Streaming library, you might be wondering how it relates to Structured Streaming. The two share the same underlying Spark engine but differ in important ways: **Spark Streaming** is based on DStreams, a sequence of RDDs produced by dividing an incoming stream into time-bounded batches. All transformations operate on RDDs, meaning you work at a lower-level API. It has limited support for event-time semantics (i.e., ordering data by when it was generated, not when it was ingested) and is no longer actively developed. **Spark Structured Streaming** is built on the DataFrame and Dataset APIs, giving you access to the full Spark SQL optimizer. It offers native event-time windowing, watermarking for handling late data, stateful aggregations, and a cleaner fault-tolerance model via checkpointing. Because it uses the same API as batch DataFrames, you can mix streaming and static data in the same job (e.g., streaming joins with a static lookup table). In short, Spark Streaming is a legacy project kept for backward compatibility, while Structured Streaming is the actively developed, recommended engine for all new streaming workloads. ## Why Integrate Bright Data’s SERP API into Spark Structured Streaming? Spark Structured Streaming provides a powerful engine for transforming and aggregating data at scale, but it needs a reliable, structured source of live web data to act on. This is where [Bright Data’s SERP API](/products/serp-api) comes in. The SERP API lets you programmatically issue queries to major search engines (including Google, Bing, DuckDuckGo, Yandex, and more) and retrieve full search engine results pages (SERPs) without being blocked. Results are returned in multiple formats: parsed JSON, a lightweight `parsed_light` variant with just the top organic results, raw HTML, or clean AI-ready Markdown. Because scraping search engines directly is notoriously difficult due to anti-bot measures, rate limits, and dynamic rendering, routing your queries through Bright Data’s infrastructure removes all of that complexity from your pipeline. Combining this with Spark Structured Streaming’s micro-batch engine creates a continuously running pipeline that periodically pulls fresh SERP data, applies transformations and aggregations at scale, and writes structured results to any sink you choose, without you having to manage proxies, CAPTCHAs, or scraping infrastructure. This approach is especially useful for: - Monitor how a set of target keywords ranks across search engines at regular intervals, write results to a Delta table, and compute rank changes over time. - Continuously fetch SERPs for competitor brand names or products, parse the structured results, and stream them into a data warehouse for dashboarding. - Poll Google News search results across multiple topics in parallel micro-batches, deduplicate articles using Spark’s stateful aggregations, and sink curated results to a data lake. - Continuously ingest SERP results to detect when paid ads appear for your target keywords, capture the ad copy and URLs, and alert downstream systems. By combining Spark Structured Streaming’s distributed, scalable processing with Bright Data’s [web access infrastructure for AI and data pipelines](/ai/web-access), you build pipelines that react to real-world search data continuously, without maintaining any scraping infrastructure of your own. ## How to Build a Continuous SERP Ingestion Pipeline with Spark Structured Streaming In this guided section, you will build a PySpark pipeline that: 1. **Triggers on a schedule** using Spark’s built-in rate source as a clock. 2. **Calls Bright Data’s SERP API** inside a `foreachBatch` function on each micro-batch to fetch live Google News results for a target topic. 3. **Parses and transforms** the structured JSON response into a clean Spark DataFrame. 4. **Writes results to a sink** (both a local JSON output directory and the console) so you can inspect the live data. > **Note:** This example demonstrates a news monitoring use case, but the same pattern applies to any continuous SERP ingestion scenario: keyword rank tracking, ad monitoring, price comparison via web search, and so on. ### Prerequisites To follow along, make sure you have: - Python 3.8+ installed. - Apache Spark 3.3+ installed locally, or access to a Databricks / AWS EMR / Google Dataproc cluster. - PySpark installed: `pip install pyspark`. - The `requests` library installed: `pip install requests`. - A Bright Data account with an active SERP API zone and API key (with Admin permissions). Follow the [official Bright Data documentation](https://docs.brightdata.com/scraping-automation/serp-api/introduction) to set up your SERP API zone and retrieve your API key. Store both your API key and zone name somewhere safe; you will need them shortly. ### Step 1: Set Up Your Project Create a new project directory and set up the files you will need: ```none mkdir spark-serp-pipeline cd spark-serp-pipeline touch pipeline.py touch config.py mkdir -p output/checkpoint ``` Open `config.py` and add your Bright Data credentials and search configuration: ```none # config.py BRIGHT_DATA_API_KEY = "YOUR_BRIGHT_DATA_API_KEY" SERP_API_ZONE = "YOUR_SERP_API_ZONE" # The search query to monitor (customize this for your use case) SEARCH_QUERY = "artificial intelligence news" # How often to trigger a new micro-batch (in seconds) TRIGGER_INTERVAL_SECONDS = 60 # Output directory for JSON results OUTPUT_PATH = "output/serp_results" CHECKPOINT_PATH = "output/checkpoint" ``` > **Security tip:** In a production environment, avoid hardcoding credentials in source files. Use environment variables, a secrets manager (e.g., AWS Secrets Manager, Azure Key Vault, HashiCorp Vault), or Databricks Secrets to inject these values at runtime. ### Step 2: Initialize the SparkSession Open `pipeline.py` and start by creating your SparkSession. This is the entry point to all Spark functionality: ```none # pipeline.py from pyspark.sql import SparkSession from pyspark.sql.types import ( StructType, StructField, StringType, IntegerType, ArrayType ) from pyspark.sql import functions as F import requests import json import config # Initialize SparkSession spark = SparkSession.builder \ .appName("BrightDataSERPStream") \ .config("spark.sql.shuffle.partitions", "4") \ .getOrCreate() # Reduce log verbosity for cleaner output spark.sparkContext.setLogLevel("WARN") print("SparkSession initialized.") ``` Setting `spark.sql.shuffle.partitions` to a small number like `4` is appropriate for a local development environment. On a cluster, you would tune this based on the size of your data and number of executor cores. ### Step 3: Define the SERP API Fetch Function Next, define the Python function that will call Bright Data’s SERP API and return parsed results. This function will be invoked from inside the Spark `foreachBatch` callback on the driver, so it uses the standard `requests` library rather than any Spark-distributed mechanism: ```none # pipeline.py (continued) def fetch_serp_results(query: str) -> list[dict]: """ Calls Bright Data's SERP API and returns a list of parsed news results. Uses the parsed_light data format for lightweight, structured JSON output. """ url = "https://api.brightdata.com/request" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {config.BRIGHT_DATA_API_KEY}" } payload = { "zone": config.SERP_API_ZONE, "url": f"https://www.google.com/search?q={query}&tbm=nws&hl=en&gl=us", "format": "raw", "data_format": "parsed_light" } try: response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() data = response.json() # The parsed_light format returns a "news" array of result objects results = data.get("news", []) print(f"[SERP API] Fetched {len(results)} results for query: '{query}'") return results except requests.exceptions.RequestException as e: print(f"[SERP API] Request failed: {e}") return [] ``` Let’s break down the key request parameters: - **`zone`**: Your SERP API zone name from the Bright Data dashboard. - **`url`**: The Google search URL. The `tbm=nws` parameter restricts results to Google News. `hl=en` sets the interface language to English, and `gl=us` targets the United States for geo-targeted results. - **`format`**: Set to `"raw"` to receive the response body directly. - **`data_format`**: Set to `"parsed_light"` to receive a clean JSON array of top organic/news results with titles, URLs, sources, and dates — without ads or knowledge panels. For full SERP data including ads and knowledge panels, use `"parsed"`. For LLM-friendly output, use `"markdown"`. ### Step 4: Build the Streaming Source Using the Rate Generator Since Spark Structured Streaming does not have a native HTTP source, we use a well-established pattern: the built-in `rate` source acts as a clock, generating one row per second (or per configured rate). Each micro-batch produced by the rate source triggers our `foreachBatch` callback, inside which we call the SERP API. Add the rate stream definition to `pipeline.py`: ```none # pipeline.py (continued) rate_stream = spark.readStream \ .format("rate") \ .option("rowsPerSecond", 1) \ .load() print("Rate stream created. Pipeline will trigger every micro-batch interval.") ``` The rate source is explicitly designed for testing and clock-driven scenarios like this one. Because real-world API rate limits apply, we will configure the trigger interval in Step 5 so that the pipeline only calls the SERP API once per minute, not once per second. ### Step 5: Define the `foreachBatch` Handler The `foreachBatch` handler is the heart of the pipeline. Spark calls this function at every micro-batch, passing in a DataFrame of that batch’s rows and a unique batch ID. Inside the function, we call the SERP API, convert the results into a Spark DataFrame, apply transformations, and write to the output sink: ```none # pipeline.py (continued) # Define the schema for parsed SERP results serp_schema = StructType([ StructField("title", StringType(), True), StructField("link", StringType(), True), StructField("source", StringType(), True), StructField("date", StringType(), True), StructField("global_rank", IntegerType(), True), ]) def process_batch(batch_df, batch_id): """ Called by Spark on each micro-batch trigger. Fetches SERP data from Bright Data, converts results to a DataFrame, and writes them to the output sink. """ print(f"\n--- Processing batch {batch_id} ---") # Fetch live SERP results from Bright Data results = fetch_serp_results(config.SEARCH_QUERY) if not results: print(f"Batch {batch_id}: No results returned. Skipping write.") return # Convert the results list to a Spark DataFrame results_df = spark.createDataFrame(results, schema=serp_schema) # Add metadata columns for tracking enriched_df = results_df \ .withColumn("query", F.lit(config.SEARCH_QUERY)) \ .withColumn("batch_id", F.lit(batch_id)) \ .withColumn("ingested_at", F.current_timestamp()) # Print to console for visibility enriched_df.show(truncate=False) # Write to JSON output (append mode, partitioned by ingestion date) enriched_df \ .withColumn("ingestion_date", F.to_date("ingested_at")) \ .write \ .mode("append") \ .partitionBy("ingestion_date") \ .json(config.OUTPUT_PATH) print(f"Batch {batch_id}: Wrote {enriched_df.count()} records to {config.OUTPUT_PATH}") ``` A few things to note about this design: **`spark.createDataFrame(results, schema=serp_schema)`** converts the Python list of dictionaries returned by the SERP API into a typed Spark DataFrame. Providing an explicit schema is preferred over schema inference — it makes the job faster and more predictable. **`F.lit(batch_id)`** attaches the current micro-batch ID to every row, which is useful for deduplication if the pipeline retries a failed batch (since `foreachBatch` provides at-least-once delivery guarantees by default). **`F.current_timestamp()`** timestamps each row with the ingestion time on the driver, giving you a reliable audit trail for when each result entered the pipeline. ### Step 6: Start the Streaming Query Now wire everything together by attaching the `foreachBatch` handler to the rate stream and starting the query: ```none # pipeline.py (continued) # Attach the foreachBatch handler and configure the trigger interval query = rate_stream.writeStream \ .foreachBatch(process_batch) \ .trigger(processingTime=f"{config.TRIGGER_INTERVAL_SECONDS} seconds") \ .option("checkpointLocation", config.CHECKPOINT_PATH) \ .start() print(f"Streaming query started. Triggering every {config.TRIGGER_INTERVAL_SECONDS} seconds.") print("Press Ctrl+C to stop.") # Wait for the query to terminate (runs indefinitely until interrupted) query.awaitTermination() ``` The `.trigger(processingTime="60 seconds")` call tells Spark to fire a new micro-batch every 60 seconds — once per minute — regardless of how many rows the rate source has generated. This is the mechanism that paces your SERP API calls, keeping you within rate limits while still running continuously. The `.option("checkpointLocation", ...)` is critical for fault tolerance. Spark writes the query’s progress metadata (offsets, committed batches) to this directory. If the process crashes and restarts, Spark reads the checkpoint to determine which batches have already been processed and resumes cleanly from the correct point. ### Step 7: Run and Inspect the Results Run the pipeline from your terminal: ```none python pipeline.py ``` You should see output similar to the following after the first trigger fires: You can see the output running on Spark at `localhost:4040`: After a few minutes of running, inspect the output directory: ```none ls output/serp_results/ ls output/serp_results/ingestion_date=2025-03-19/ ``` You can read the results back into Spark for ad-hoc analysis at any time: ```none # Read back the accumulated results df = spark.read.json("output/serp_results/") df.orderBy("ingested_at", ascending=False).show(20, truncate=False) ``` [Here](https://github.com/icode247/spark-serp-pipeline) is the full pipeline code in one place for easy reference. ## Taking It Further This example demonstrates a foundational ingestion pattern, but there are many directions you can take it: - Instead of one topic, maintain a list of keywords and fan out to parallel SERP API calls inside each `foreachBatch` invocation. Use Python’s `concurrent.futures.ThreadPoolExecutor` to call the API for multiple queries simultaneously within the same micro-batch. - Replace the JSON sink with a Delta table for ACID-compliant, incremental writes with schema evolution support. This makes historical querying and deduplication much simpler. - Bright Data’s SERP API supports [Bing search engine queries](https://brightdata.com/products/serp-api/bing-search) alongside Google, DuckDuckGo, Yandex, and more. Poll multiple engines in parallel within the same batch and merge the result sets. - Use Bright Data’s [Web Unlocker](https://brightdata.com/products/web-unlocker) to follow the URLs returned by the SERP API and retrieve the full HTML or Markdown content of each article. Pipe that content into a downstream NLP stage within the same Spark pipeline. - Deploy the pipeline on Databricks, AWS EMR, or Google Dataproc for production-grade scalability. On Databricks, you can also use Delta Live Tables to manage the pipeline declaratively. - Write the enriched SERP results to a Kafka topic and consume them in real time from downstream microservices, dashboards, or alert systems. ## Conclusion In this tutorial, you learned how to use Bright Data’s SERP API to continuously ingest live search engine results and process them with Apache Spark Structured Streaming. Using the `rate` source as a scheduling clock and `foreachBatch` as the integration bridge, you built a continuously running pipeline that fetches fresh SERP data on every trigger, transforms it into a typed Spark DataFrame, and writes results to a partitioned JSON sink, all with fault-tolerant checkpointing built in. This pattern is ideal for any team that needs to process real-time web search signals at scale: keyword rank tracking, competitive monitoring, news aggregation, ad intelligence, and more. Unlike ad-hoc script-based polling, a Spark Structured Streaming pipeline gives you a distributed, recoverable, and easily extensible foundation that grows with your data volumes. To build more advanced pipelines, explore Bright Data’s [full suite of web data products](https://brightdata.com/products), including the Web Unlocker for bypassing bot protection on arbitrary URLs, the Scraping Browser for JavaScript-heavy sites, and ready-made datasets for the most popular platforms. [Sign up for a free Bright Data account](https://brightdata.com/) today and start powering your data pipelines with reliable, real-time web data. Contact usStart free trial No credit card required [ ](https://www.linkedin.com/in/arindam2004/) Arindam Majumder Technical Writer Arindam Majumder is a developer advocate, YouTuber, and technical writer who simplifies LLMs, agent workflows, and AI content for 5,000+ followers. Expertise RAG AI Agents Python [ View all articles ](https://brightdata.com/blog/authors/arindam-majumder) Table of Contents Dedicated Scraper APIs & No-Code Scrapers Over 1000 scrapers for all popular domains. Simplify your web scraping. [See pricing](/pricing/web-scraper "See pricing") Just want data? Skip scraping. Hundreds of ready-to-use datasets from all popular domains. [See pricing](/pricing/datasets "See pricing") [ ](https://news.ycombinator.com/submitlink?t=Spark+Structured+Streaming+with+continuous+web+data+ingestion&u=https://brightdata.com/blog/web-data/spark-structured-streaming-with-web-data) [ ](https://www.linkedin.com/shareArticle?mini=true&title=Spark+Structured+Streaming+with+continuous+web+data+ingestion&url=https://brightdata.com/blog/web-data/spark-structured-streaming-with-web-data) [ ](http://www.reddit.com/submit?title=Spark+Structured+Streaming+with+continuous+web+data+ingestion&url=https://brightdata.com/blog/web-data/spark-structured-streaming-with-web-data) ## You might also be interested in [ ](https://brightdata.com/blog/ai/openhuman-with-bright-data "Production-Ready Web Access in OpenHuman Through the Bright Data CLI") [AI Antonello Zanini Technical Writer ### Production-Ready Web Access in OpenHuman Through the Bright Data CLI Integrate Bright Data CLI with OpenHuman to enable production-ready web access and data collection for AI agents. 09-Sep-2026 12 min read ](https://brightdata.com/blog/ai/openhuman-with-bright-data) [ ](https://brightdata.com/blog/ai/minimax-m3-with-bright-data "Giving self-hosted MiniMax M3 agents live web access with Bright Data") [AI Satyam Tripathi Technical Writer ### Giving self-hosted MiniMax M3 agents live web access with Bright Data Self-hosted MiniMax M3 agents get live web access using Bright Data’s search and scraping tools. Bypass blocks and CAPTCHAs. 09-Sep-2026 54 min read ](https://brightdata.com/blog/ai/minimax-m3-with-bright-data) [ ](https://brightdata.com/blog/web-data/multimodal-web-scraping-with-minimax "Multimodal Web Scraping with MiniMax") [Web Data Antonello Zanini Technical Writer ### Multimodal Web Scraping with MiniMax Pair Bright Data Web Unlocker with MiniMax M3 vision to extract structured data from images and web page screenshots. 09-Sep-2026 4 min read ](https://brightdata.com/blog/web-data/multimodal-web-scraping-with-minimax)