Bright Data의 SERP API로 구조화된 스트리밍을 시작하세요

이 글에서는 다음 내용을 다룹니다:

Apache Spark Structured Streaming이 무엇이며 어떤 기능을 제공하는지.
Bright Data의 SERP API를 Spark Structured Streaming 파이프라인에 통합하는 것이 왜 성공적인 전략인지.
Bright Data의 SERP API를 사용하여 실시간 웹 검색 데이터를 지속적으로 수집하는 PySpark 파이프라인을 구축하는 방법.

그럼 시작해 봅시다!

Apache Spark Structured Streaming이란 무엇인가요?

Apache Spark Structured Streaming은 Spark SQL 엔진을 기반으로 구축된 확장 가능하고 내결함성이 뛰어난 스트림 처리 엔진입니다. (DStreams를 사용하여 데이터를 개별 RDD 기반 마이크로 배치로 분할하는) 구형 Spark Streaming 라이브러리와 달리, Structured Streaming은 실시간 데이터 스트림을 지속적으로 추가되는 무제한 테이블로 취급합니다. 정적 배치 작업에서 작성하는 것과 동일한 DataFrame 및 SQL API 코드를 작성하면, Spark가 새로운 데이터가 도착함에 따라 이를 증분식으로 실행합니다.

이 엔진은 기본적으로 마이크로 배치 실행 모델로 작동합니다. 각 트리거 간격마다 Spark는 소스에서 최신 데이터를 읽어 처리한 후 결과를 싱크에 기록합니다. 체크포인팅을 통해 진행 상황을 추적하므로, 파이프라인은 장애 발생 시 복구되어 중단된 지점부터 정확히 재개될 수 있어 종단 간 내결함성을 보장합니다.

Structured Streaming은 Kafka 토픽, Delta 테이블, Auto Loader를 통한 클라우드 오브젝트 스토리지, 레이트 제너레이터(테스트용) 등 다양한 내장 소스를 지원합니다. 기본적으로 지원되지 않는 소스(예: REST API)의 경우, 각 마이크로 배치(micro-batch)를 Python 함수에 전달하여 임의의 인제스트 로직을 구현할 수 있는 foreachBatch 확장 메서드를 사용할 수 있습니다. 여기서는 이 접근 방식을 사용할 것입니다.

Spark Streaming 대 Spark Structured Streaming: 차이점은 무엇인가?

기존 Spark Streaming 라이브러리에 익숙한 분이라면, 이것이 Structured Streaming과 어떤 관련이 있는지 궁금할 수 있습니다. 두 라이브러리는 동일한 기본 Spark 엔진을 공유하지만 중요한 측면에서 차이가 있습니다:

Spark Streaming은 들어오는 스트림을 시간 기반 배치로 나누어 생성된 RDD 시퀀스인 DStreams를 기반으로 합니다. 모든 변환은 RDD를 대상으로 수행되므로, 사용자는 저수준 API에서 작업하게 됩니다. 이벤트 시간 세미오틱(즉, 데이터가 수집된 시점이 아닌 생성된 시점에 따라 데이터를 정렬하는 방식)에 대한 지원이 제한적이며, 더 이상 활발하게 개발되지 않고 있습니다.

Spark Structured Streaming은 DataFrame 및 Dataset API를 기반으로 구축되어, Spark SQL 최적화 기능을 모두 활용할 수 있습니다. 이 기능은 네이티브 이벤트 타임 윈도잉, 지연 데이터 처리를 위한 워터마킹, 상태 기반 집계, 그리고 체크포인팅을 통한 더 간결한 내결함성 모델을 제공합니다. 배치 DataFrame과 동일한 API를 사용하므로, 동일한 작업 내에서 스트리밍 데이터와 정적 데이터를 혼합하여 사용할 수 있습니다(예: 정적 조회 테이블을 사용한 스트리밍 조인).

간단히 말해, Spark Streaming은 하위 호환성을 위해 유지되는 레거시 프로젝트인 반면, Structured Streaming은 모든 새로운 스트리밍 워크로드를 위해 적극적으로 개발되고 권장되는 엔진입니다.

왜 Bright Data의 SERP API를 Spark Structured Streaming에 통합해야 할까요?

Spark Structured Streaming은 대규모 데이터 변환 및 집계 작업을 위한 강력한 엔진을 제공하지만, 이를 실행하기 위해서는 신뢰할 수 있고 구조화된 실시간 웹 데이터 소스가 필요합니다. 바로 여기서 Bright Data의 SERP API가 그 역할을 합니다.

SERP API를 사용하면 주요 검색 엔진(Google, Bing, DuckDuckGo, Yandex 등 포함)에 프로그래밍 방식으로 쿼리를 전송하고 차단되지 않은 상태로 전체 검색 결과 페이지(SERP)를 가져올 수 있습니다. 결과는 파싱된 JSON, 상위 자연 검색 결과만 포함된 경량화된 parsed_light 변형, 원시 HTML, 또는 AI 처리에 최적화된 깔끔한 마크다운 등 다양한 형식으로 반환됩니다. 봇 방지 조치, 속도 제한, 동적 렌더링 등으로 인해 검색 엔진을 직접 스크래핑하는 것은 매우 어렵기로 유명하지만, Bright Data의 인프라를 통해 쿼리를 전송하면 파이프라인에서 이러한 모든 복잡성을 제거할 수 있습니다.

이를 Spark Structured Streaming의 마이크로 배치 엔진과 결합하면, 프록시, CAPTCHA 또는 스크래핑 인프라를 직접 관리할 필요 없이 주기적으로 최신 SERP 데이터를 가져오고, 대규모 변환 및 집계 작업을 수행하며, 선택한 싱크에 구조화된 결과를 기록하는 지속적으로 실행되는 파이프라인을 구축할 수 있습니다.

이 접근 방식은 특히 다음 용도에 유용합니다:

정기적으로 특정 타겟 키워드 세트의 검색 엔진별 순위를 모니터링하고, 결과를 델타 테이블에 기록한 후, 시간 경과에 따른 순위 변화를 계산합니다.
경쟁사 브랜드명이나 제품에 대한 SERP를 지속적으로 가져와 구조화된 결과를 파싱하고, 대시보드 생성을 위해 데이터 웨어하우스로 스트리밍합니다.
여러 주제에 대한 Google 뉴스 검색 결과를 병렬 마이크로 배치로 폴링하고, Spark의 상태 기반 집계 기능을 사용하여 중복 기사를 제거한 후, 선별된 결과를 데이터 레이크로 전송합니다.
SERP 결과를 지속적으로 수집하여 대상 키워드에 대한 유료 광고가 표시되는 시점을 감지하고, 광고 문구와 URL을 캡처한 후 하류 시스템에 알림을 전송합니다.

Spark Structured Streaming의 분산형 확장 가능 처리 기능과 AI 및 데이터 파이프라인을 위한 Bright Data의 웹 액세스 인프라를 결합하면, 자체 스크래핑 인프라를 유지 관리할 필요 없이 실제 검색 데이터에 지속적으로 반응하는 파이프라인을 구축할 수 있습니다.

Spark Structured Streaming을 사용하여 지속적인 SERP 수집 파이프라인 구축하는 방법

이 안내 섹션에서는 다음과 같은 기능을 수행하는 PySpark 파이프라인을 구축하게 됩니다.

Spark의 내장된 레이트 소스를 시계로 사용하여일정대로 트리거됩니다.
각 마이크로 배치의 foreachBatch 함수 내에서Bright Data의 SERP API를 호출하여 대상 주제에 대한 실시간 Google 뉴스 결과를 가져옵니다.
구조화된 JSON 응답을파싱하고 변환하여 깔끔한 Spark DataFrame으로 만듭니다.
실시간 데이터를 확인할 수 있도록결과를 싱크 (로컬 JSON 출력 디렉터리와 콘솔 모두)에 기록합니다.

참고: 이 예제는 뉴스 모니터링 사용 사례를 보여줍니다만, 동일한 패턴은 키워드 순위 추적, 광고 모니터링, 웹 검색을 통한 가격 비교 등 모든 지속적인 SERP 수집 시나리오에 적용됩니다.

필수 사항

이 예제를 따라 하려면 다음이 설치되어 있는지 확인하십시오:

Python 3.8+가 설치되어 있어야 합니다.
로컬에 Apache Spark 3.3 이상이 설치되어 있거나, Databricks / AWS EMR / Google Dataproc 클러스터에 액세스할 수 있어야 합니다.
PySpark 설치: pip install pyspark.
requests 라이브러리 설치: pip install requests.
활성 SERP API 영역과 API 키(관리자 권한 포함)가 있는 Bright Data 계정.

공식 Bright Data 문서를 따라 SERP API 영역을 설정하고 API 키를 가져오십시오. API 키와 영역 이름을 안전한 곳에 보관하십시오. 곧 필요하게 될 것입니다.

1단계: 프로젝트 설정

새 프로젝트 디렉터리를 생성하고 필요한 파일을 설정합니다:

mkdir spark-serp-pipeline
cd spark-serp-pipeline
touch pipeline.py
touch config.py
mkdir -p output/checkpoint

config.py 파일을 열고 Bright Data 자격 증명 및 검색 구성을 추가하세요:

# config.py

BRIGHT_DATA_API_KEY = "YOUR_BRIGHT_DATA_API_KEY"
SERP_API_ZONE = "YOUR_SERP_API_ZONE"

# 모니터링할 검색 쿼리 (사용 사례에 맞게 수정하세요)
SEARCH_QUERY = "artificial intelligence news"

# 새 마이크로 배치 트리거 간격 (초 단위)
TRIGGER_INTERVAL_SECONDS = 60

# JSON 결과 출력 디렉터리
OUTPUT_PATH = "output/serp_results"
CHECKPOINT_PATH = "output/checkpoint"

보안 팁: 프로덕션 환경에서는 소스 파일에 자격 증명을 하드코딩하지 마십시오. 환경 변수, 시크릿 관리자(예: AWS Secrets Manager, Azure Key Vault, HashiCorp Vault) 또는 Databricks Secrets를 사용하여 런타임에 이러한 값을 주입하십시오.

2단계: SparkSession 초기화

pipeline.py 파일을 열고 SparkSession을 생성하는 것으로 시작합니다. 이는 모든 Spark 기능의 진입점입니다:

# 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

# SparkSession 초기화
spark = SparkSession.builder 
    .appName("BrightDataSERPStream") 
    .config("spark.sql.shuffle.partitions", "4") 
    .getOrCreate()

# 더 깔끔한 출력을 위해 로그 상세 수준 낮추기
spark.sparkContext.setLogLevel("WARN")

print("SparkSession 초기화됨.")

spark.sql.shuffle.partitions를 4와 같이 작은 숫자로 설정하는 것은 로컬 개발 환경에 적합합니다. 클러스터 환경에서는 데이터의 크기와 익스큐터 코어 수에 따라 이 값을 조정해야 합니다.

3단계: SERP API 가져오기 함수 정의

다음으로, Bright Data의 SERP API를 호출하고 파싱된 결과를 반환할 Python 함수를 정의합니다. 이 함수는 드라이버의 Spark foreachBatch 콜백 내부에서 호출되므로, Spark의 분산 메커니즘 대신 표준 requests 라이브러리를 사용합니다:

# pipeline.py (계속)

def fetch_serp_results(query: str) -> list[dict]:
    """
    Bright Data의 SERP API를 호출하고 파싱된 뉴스 결과 목록을 반환합니다.
    가볍고 구조화된 JSON 출력을 위해 parsed_light 데이터 형식을 사용합니다.
    """
    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()

        # parsed_light 형식은 결과 객체들의 "news" 배열을 반환합니다
        results = data.get("news", [])
        print(f"[SERP API] 쿼리 '{query}'에 대한 {len(results)}개의 결과를 가져왔습니다.")
        return results

    except requests.exceptions.RequestException as e:
        print(f"[SERP API] 요청 실패: {e}")
        return []

주요 요청 매개변수를 자세히 살펴보겠습니다:

zone: Bright Data 대시보드에서 확인한 SERP API 영역 이름입니다.
url: Google 검색 URL입니다. tbm=nws 매개변수는 결과를 Google 뉴스로 제한합니다. hl=en은 인터페이스 언어를 영어로 설정하고, gl=us는 지역 타겟팅 결과를 위해 미국을 대상으로 지정합니다.
format: 응답 본문을 직접 수신하려면 "raw" 로 설정합니다.
data_format: "parsed_light" 로 설정하면 광고나 지식 패널 없이 제목, URL, 출처, 날짜가 포함된 상위 자연 검색/뉴스 결과의 깔끔한 JSON 배열을 수신할 수 있습니다. 광고와 지식 패널을 포함한 전체 SERP 데이터를 얻으려면 "parsed"를 사용하세요. LLM에 적합한 출력을 원한다면 "markdown"을 사용하세요.

4단계: 레이트 생성기를 사용하여 스트리밍 소스 구축

Spark Structured Streaming에는 기본 HTTP 소스가 없으므로, 잘 정립된 패턴을 사용합니다. 내장된 레이트 소스가 시계 역할을 하여 초당(또는 구성된 속도에 따라) 한 행을 생성합니다. 레이트 소스가 생성하는 각 마이크로 배치마다 foreachBatch 콜백이 트리거되며, 이 콜백 내부에서 SERP API를 호출합니다.

pipeline.py에 레이트 스트림 정의를 추가합니다:

# pipeline.py (계속)
rate_stream = spark.readStream 
    .format("rate") 
    .option("rowsPerSecond", 1) 
    .load()

print("레이트 스트림이 생성되었습니다. 파이프라인은 매 마이크로 배치 간격마다 트리거됩니다.")

이 레이트 소스는 이와 같은 테스트 및 시계 기반 시나리오를 위해 특별히 설계되었습니다. 실제 API의 속도 제한이 적용되므로, 5단계에서 트리거 간격을 구성하여 파이프라인이 초당 한 번이 아닌 분당 한 번만 SERP API를 호출하도록 할 것입니다.

5단계: `foreachBatch` 핸들러 정의

foreachBatch 핸들러는 파이프라인의 핵심입니다. Spark는 매 마이크로 배치마다 이 함수를 호출하며, 해당 배치의 행으로 구성된 DataFrame과 고유한 배치 ID를 전달합니다. 함수 내부에서는 SERP API를 호출하고, 결과를 Spark DataFrame으로 변환한 후, 변환을 적용하고 출력 싱크에 기록합니다:

# pipeline.py (계속)

# 파싱된 SERP 결과의 스키마 정의
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):
    """
    각 마이크로 배치 트리거 시 Spark에 의해 호출됩니다.
    Bright Data에서 SERP 데이터를 가져와 결과를 DataFrame으로 변환한 후,
    출력 싱크에 기록합니다.
    """
    print(f"n--- 배치 {batch_id} 처리 중 ---")

    # Bright Data에서 실시간 SERP 결과 가져오기
    results = fetch_serp_results(config.SEARCH_QUERY)

    if not results:
        print(f"배치 {batch_id}: 결과가 반환되지 않았습니다. 쓰기 건너뜁니다.")
        return

    # 결과 목록을 Spark DataFrame으로 변환
    results_df = spark.createDataFrame(results, schema=serp_schema)

    # 추적을 위한 메타데이터 열 추가
    enriched_df = results_df 
        .withColumn("query", F.lit(config.SEARCH_QUERY)) 
        .withColumn("batch_id", F.lit(batch_id)) 
        .withColumn("ingested_at", F.current_timestamp())

    # 확인을 위해 콘솔에 출력
    enriched_df.show(truncate=False)

    # JSON 출력 파일로 쓰기 (추가 모드, 수집 날짜별로 파티션)
    enriched_df 
        .withColumn("ingestion_date", F.to_date("ingested_at")) 
        .write 
        .mode("append") 
        .partitionBy("ingestion_date") 
        .json(config.OUTPUT_PATH)

    print(f"배치 {batch_id}: {enriched_df.count()} 개의 레코드를 {config.OUTPUT_PATH}에 기록함")

이 설계에 대해 유의해야 할 몇 가지 사항:

spark.createDataFrame(results, schema=serp_schema)는 SERP API에서 반환된 Python 사전 리스트를 타입이 지정된 Spark DataFrame으로 변환합니다. 스키마 추론보다는 명시적인 스키마를 제공하는 것이 더 바람직합니다. 이렇게 하면 작업 속도가 빨라지고 예측 가능성도 높아집니다.

F.lit(batch_id) 는 모든 행에 현재 마이크로 배치 ID를 연결합니다. 이는 파이프라인이 실패한 배치를 재시도할 때 중복 제거에 유용합니다( foreachBatch는 기본적으로 최소 1회 전달을 보장하기 때문입니다).

F.current_timestamp() 는 각 행에 드라이버의 인제스트 시간을 타임스탬프로 기록하여, 각 결과가 파이프라인에 입력된 시점에 대한 신뢰할 수 있는 감사 추적을 제공합니다.

6단계: 스트리밍 쿼리 시작

이제 rate 스트림에 foreachBatch 핸들러를 연결하고 쿼리를 시작하여 모든 요소를 연결합니다:

# pipeline.py (계속)

# foreachBatch 핸들러 연결 및 트리거 간격 구성
query = rate_stream.writeStream 
    .foreachBatch(process_batch) 
    .trigger(processingTime=f"{config.TRIGGER_INTERVAL_SECONDS} 초") 
    .option("checkpointLocation", config.CHECKPOINT_PATH) 
    .start()

print(f"스트리밍 쿼리가 시작되었습니다. {config.TRIGGER_INTERVAL_SECONDS}초마다 트리거됩니다.")
print("중지하려면 Ctrl+C를 누르세요.")

# 쿼리가 종료될 때까지 대기합니다(중단될 때까지 무한정 실행됨)
query.awaitTermination()

.trigger(processingTime="60 seconds") 호출은 레이트 소스가 생성한 행 수와 관계없이 60초마다(분당 1회) 새로운 마이크로 배치를 실행하도록 Spark에 지시합니다. 이는 SERP API 호출의 속도를 조절하여, 지속적으로 실행하면서도 레이트 제한을 준수하도록 하는 메커니즘입니다.

.option("checkpointLocation", ...) 은 장애 허용에 매우 중요합니다. Spark는 쿼리의 진행 상황 메타데이터(오프셋, 커밋된 배치)를 이 디렉터리에 기록합니다. 프로세스가 중단되었다가 다시 시작되면, Spark는 체크포인트를 읽어 이미 처리된 배치를 파악하고 올바른 지점에서 원활하게 작업을 재개합니다.

7단계: 실행 및 결과 확인

터미널에서 파이프라인을 실행합니다:

python pipeline.py

첫 번째 트리거가 실행된 후 다음과 유사한 출력이 표시됩니다:

localhost:4040에서 실행 중인 출력을 확인할 수 있습니다:

Spark UI showing the streaming query plan with the "rate" source node visible

몇 분간 실행한 후 출력 디렉터리를 확인해 보세요:

ls output/serp_results/
ls output/serp_results/ingestion_date=2025-03-19/

언제든지 결과를 Spark로 다시 불러와 즉석 분석을 수행할 수 있습니다:

# 누적된 결과 다시 불러오기
df = spark.read.json("output/serp_results/")
df.orderBy("ingested_at", ascending=False).show(20, truncate=False)

Terminal showing the console output of the enriched DataFrame with SERP results populated in the columns

참조하기 쉽도록 전체 파이프라인 코드를 한 곳에모아두었습니다.

한 단계 더 나아가기

이 예제는 기본적인 데이터 수집 패턴을 보여줍니다만, 이를 확장할 수 있는 방법은 다양합니다:

단일 토픽 대신 키워드 목록을 유지하고, 각 foreachBatch 호출 내에서 SERP API 호출을 병렬로 분산(fan out)합니다. Python의 concurrent.futures.ThreadPoolExecutor를 사용하여 동일한 마이크로 배치 내에서 여러 쿼리에 대한 API를 동시에 호출합니다.
JSON 싱크를 Delta 테이블로 교체하여 스키마 진화 기능을 지원하는 ACID 준수 증분 쓰기를 구현할 수 있습니다. 이를 통해 과거 쿼리 조회 및 중복 제거가 훨씬 간편해집니다.
Bright Data의 SERP API는 Google, DuckDuckGo, Yandex 등 외에도 Bing 검색 엔진 쿼리를 지원합니다. 동일한 배치 내에서 여러 검색 엔진을 병렬로 폴링하고 결과 집합을 병합하세요.
Bright Data의 Web Unlocker를 사용하여 SERP API에서 반환된 URL을 추적하고 각 기사의 전체 HTML 또는 Markdown 콘텐츠를 가져옵니다. 해당 콘텐츠를 동일한 Spark 파이프라인 내의 다운스트림 NLP 단계로 전달합니다.
Databricks, AWS EMR 또는 Google Dataproc에 파이프라인을 배포하여 프로덕션급 확장성을 확보하세요. Databricks에서는 Delta Live Tables를 사용하여 파이프라인을 선언적으로 관리할 수도 있습니다.
보강된 SERP 결과를 Kafka 토픽에 기록하고, 다운스트림 마이크로서비스, 대시보드 또는 알림 시스템에서 실시간으로 소비합니다.

결론

이 튜토리얼에서는 Bright Data의 SERP API를 사용하여 실시간 검색 엔진 결과를 지속적으로 수집하고 Apache Spark Structured Streaming으로 처리하는 방법을 배웠습니다. 레이트 소스를 스케줄링 클럭으로, foreachBatch를 통합 브리지로 사용하여, 트리거될 때마다 최신 SERP 데이터를 가져와 타입이 지정된 Spark DataFrame으로 변환하고, 결과를 파티션화된 JSON 싱크에 기록하는, 모두 내결함성 체크포인트 기능이 내장된 지속적으로 실행되는 파이프라인을 구축했습니다.

이 패턴은 키워드 순위 추적, 경쟁사 모니터링, 뉴스 집계, 광고 인텔리전스 등 대규모 실시간 웹 검색 신호를 처리해야 하는 모든 팀에 이상적입니다. 임시 스크립트 기반 폴링과 달리, Spark Structured Streaming 파이프라인은 데이터 양의 증가에 따라 확장 가능한 분산형, 복구 가능한 기반을 제공합니다.

보다 고급 파이프라인을 구축하려면, 임의의 URL에서 봇 보호 기능을 우회하는 Web Unlocker, 자바스크립트 기반 사이트용 Scraping Browser, 그리고 가장 인기 있는 플랫폼을 위한 기성 데이터셋을 포함한 Bright Data의 웹 데이터 제품군 전체를 살펴보세요.

지금 바로Bright Data 무료 계정에 가입하고, 신뢰할 수 있는 실시간 웹 데이터로 데이터 파이프라인을 강화해 보세요.

문의하기 무료 체험 시작하기

지속적인 웹 데이터 수집을 통한 Spark Structured Streaming

Apache Spark Structured Streaming이란 무엇인가요?

Spark Streaming 대 Spark Structured Streaming: 차이점은 무엇인가?

왜 Bright Data의 SERP API를 Spark Structured Streaming에 통합해야 할까요?

Spark Structured Streaming을 사용하여 지속적인 SERP 수집 파이프라인 구축하는 방법

필수 사항

1단계: 프로젝트 설정

2단계: SparkSession 초기화

3단계: SERP API 가져오기 함수 정의

4단계: 레이트 생성기를 사용하여 스트리밍 소스 구축

5단계: `foreachBatch` 핸들러 정의

6단계: 스트리밍 쿼리 시작

7단계: 실행 및 결과 확인

한 단계 더 나아가기

결론

관심이 있을 수 있습니다.

프롬프트로 Kiro와 Bright Data를 활용한 웹 스크레이퍼 구축

Bright Data로 Oracle Generative AI 에이전트에 웹 접근 권한 부여하기

goose를 Bright Data의 Web MCP에 연결하는 방법

지속적인 웹 데이터 수집을 통한 Spark Structured Streaming

Apache Spark Structured Streaming이란 무엇인가요?

Spark Streaming 대 Spark Structured Streaming: 차이점은 무엇인가?

왜 Bright Data의 SERP API를 Spark Structured Streaming에 통합해야 할까요?

Spark Structured Streaming을 사용하여 지속적인 SERP 수집 파이프라인 구축하는 방법

필수 사항

1단계: 프로젝트 설정

2단계: SparkSession 초기화

3단계: SERP API 가져오기 함수 정의

4단계: 레이트 생성기를 사용하여 스트리밍 소스 구축

5단계: foreachBatch 핸들러 정의

6단계: 스트리밍 쿼리 시작

7단계: 실행 및 결과 확인

한 단계 더 나아가기

결론

관심이 있을 수 있습니다.

프롬프트로 Kiro와 Bright Data를 활용한 웹 스크레이퍼 구축

Bright Data로 Oracle Generative AI 에이전트에 웹 접근 권한 부여하기

goose를 Bright Data의 Web MCP에 연결하는 방법

5단계: `foreachBatch` 핸들러 정의