ScrapelessとSnowpipe Streamingを使用してWebデータをSnowflakeにストリーミングする方法

主なポイント：

• 固定スキーマなしでSnowflakeにウェブデータを取り込む。 Scrapeless Scraping Browserは、クラウドブラウザーでページをレンダリングし、改行区切りのJSON（NDJSON）を出力；SnowflakeはこれをVARIANT列に取り込み、新しいフィールドがロードを破壊することはない。
• 4つの取り込み方法、1つのデータ形状。 一回限りのロードのための一括COPY INTO、連続ファイルベースのバッチ用のSnowpipe、低遅延の行用のSnowpipe Streaming、イベント駆動のパイプライン用のKafkaコネクタ — すべて、Scrapelessが生成する同じNDJSONを読み取る。
• Snowpipe Streamingの高性能アーキテクチャは一般提供中（2025年9月からGA）。 SDK（Java、Python、Node.js）またはRESTを通じて直接行を書き込み、チャネル、オフセットトークン、正確に1回の回復を行う — ステージファイルなし。
• Schema-on-readによりスクレイピングデータを柔軟に保つ。 VARIANT列をcol:field::type表記でクエリし、LATERAL FLATTENで配列を解除 — ソースページがフィールドを追加しても移行は不要。
• Snowflake CLI（snow）は現在のツール。 pip install snowflake-cli、その後雪 sql -f ingest.sqlで1つのファイルから全てのセットアップを実行。
• 無料で開始。 新しいScrapelessアカウントには無料のScraping Browserランタイムが含まれている — Scrapeless Websiteでサインアップ。

はじめに：レンダリングされたページからSnowflakeテーブルへ

分析チームはますますウェブデータ — 製品カタログ、リスト、レビュー、市場シグナル — をファーストパーティデータと同じ倉庫に取り込みたいと考えており、それによって結合、モデリング、BIに供給できるようにしています。Snowflakeは、VARIANT型がセミ構造化JSONをネイティブに保存し、SQLでクエリ可能にするため、一般的な行き先です。

摩擦は二つのシステムの間にあります。スクレイピングされたページはJavaScriptでレンダリングされ、アンチボット防御の背後にあります；データはしばしばネストされたJSONとして到着し、ソースサイトが変化するにつれてその形状が変わります。すべてのフィールドを列にマッピングする手作りのローダーは、ページがフィールドを1つ追加するたびに壊れます。

この記事では、そのギャップを閉じるターミナルファーストのワークフローを説明します。Scrapeless Scraping Browserはレンダリングとアンチ検出を処理し、NDJSONを出力します；Snowflakeは、データの新鮮さに応じて異なる4つの方法でこれを取り込みます。例のプロデューサーは公共のスクレイピングサンドボックスbooks.toscrape.comであり、以下のすべてのコマンドは再現可能です — 同じパターンはより難しいターゲットにも適用されます（関連する2026年のベストZillowスクレイパーおよび2026年のベストAmazonスクレイパーガイドを参照）。

できること

• ウェブデータレイクハウスを構築。 スクレイピングしたカタログとリストをSnowflakeに取り込み、内部の販売や在庫データと結合。
• 定期的な市場スナップショットを実行。 実行ごとに新しいNDJSONファイルをステージに投入し、数分でSnowpipeが自動的にロード。
• ほぼリアルタイムのダッシュボードにデータを供給。 Snowpipe Streamingを使用してスクレイピングされたイベントを行単位でストリームし、1分未満の新鮮さを維持。
• 既存のKafkaバックボーンを橋渡し。 スクレイピングされたレコードをトピックにプッシュし、Snowflake Kafkaコネクタがそれらを取り込む。
• スキーマを柔軟に保つ。 生のJSONをVARIANTに保存し、クエリ時に形を整えるので、ソースサイトの変更がロードを妨げることはない。

Scrapelessでは、適用される法律、規制、およびウェブサイトのプライバシーポリシーを厳密に遵守しながら、公開されているデータのみにアクセスします。この投稿の内容はデモ目的のみです。

なぜScrapeless Scraping Browserか

Scrapeless Scraping Browserは、ウェブクローラーとAIエージェント向けに設計されたカスタマイズ可能な、アンチ検出型のクラウドブラウザーです。Snowflakeパイプラインのプロデューサー側として、以下を提供します：

• データが抽出前にDOMに存在するクラウド側のJavaScriptレンダリング
• 195以上の国における住宅用プロキシ、セッションごとに固定
• アンチ検出用ブラウザーフィンガープリンティング
• 一つのscrapeless-scraping-browser CLIインターフェイス、そのevalが1ステップでNDJSONに再形成可能なJSONを返す
• マルチページクローリングのためのセッションの持続性

無料プランでAPIキーを取得するには、Scrapeless Websiteを訪れてください。

前提条件

• Node.js 18以上
• ScrapelessアカウントとAPIキー — Scrapeless Websiteでサインアップ
• データベース、ウェアハウス、ステージ、パイプを作成できる役割を持つSnowflakeアカウント
• Snowflake CLI: pip install snowflake-cli（Python 3.10以上）
• Snowpipe自動取り込みおよび外部ステージ用：クラウドバケット（AWS S3、GCS、またはAzure）およびストレージインテグレーションを作成する権限
• jqは任意（NDJSON変換用にNodeのワンライナーのフォールバックが示されています）

一度、~/.snowflake/config.tomlでSnowflake接続を設定します：

[connections.demo]
account   = "myorg-myaccount"
user      = "jondoe"

password = "あなたのパスワードをここに"
warehouse = "ingest_wh"
database = "web_data"
schema = "raw"
role = "sysadmin"

次に、`snow sql -c demo -q "SELECT CURRENT_VERSION();"`で動作を確認します。

---

## パイプラインの概要

Scrapeless Scraping Browser → NDJSONファイル → Snowflakeステージ → テーブル (VARIANT)
(レンダリング + 抽出) (1行ごとに1つのオブジェクト) (内部または外部バケット) COPY INTO | 一時的
Snowpipe | 継続的
ストリーミング | 低遅延
Kafka | イベント駆動

形状は変わりません：Scrapelessは1行ごとに1つのJSONオブジェクトを出力し、ファイルはステージに配置され、4つの方法のいずれかで`VARIANT`カラムにロードされ、SQLでクエリされます。

---

## ステップ1 — ScrapelessでNDJSONを生成する

CLIをインストールし、キーを設定します：

```bash
npm install -g scrapeless-scraping-browser
scrapeless-scraping-browser config set apiKey あなたのAPIトークンをここに

1つのクラウドセッションを開き、カタログページに移動し、安定したマーカーを待ってから、evalで書籍レコードを抽出します。new-session JSONはdata.taskIdの下にIDをネストします — jqを使用するか、以下のポータブルgrepフォールバックを示します：

# セッションを開き、タスクIDをキャプチャします（jqパスは.data.taskId）
SID=$(scrapeless-scraping-browser new-session --name books --ttl 300 --proxy-country US --json | jq -r '.data.taskId')
# jqは使用しない？ポータブルフォールバック：
# SID=$(scrapeless-scraping-browser new-session --name books --ttl 300 --proxy-country US --json | grep -oE '"taskId":"[^"]*"' | head -1 | cut -d'"' -f4)

# カタログページをレンダリングし、次に製品グリッドを待ちます
scrapeless-scraping-browser --session-id "$SID" open "https://books.toscrape.com/catalogue/page-1.html"
scrapeless-scraping-browser --session-id "$SID" wait "article.product_pod"

# 書籍ごとに1つのレコードを抽出します；evalはJSON配列を返します
scrapeless-scraping-browser --session-id "$SID" eval '
  JSON.stringify(Array.from(document.querySelectorAll("article.product_pod")).map(el => ({
    title: el.querySelector("h3 a")?.getAttribute("title") ?? null,
    price: el.querySelector(".price_color")?.textContent.trim() ?? null,
    rating: el.querySelector("p.star-rating")?.className.replace("star-rating", "").trim() ?? null,
    in_stock: /In stock/i.test(el.querySelector(".availability")?.textContent ?? ""),
    url: el.querySelector("h3 a")?.href ?? null
  })))
' > books.raw.json

scrapeless-scraping-browser --session-id "$SID" close

配列をNDJSONに変換します — 1行ごとに1つのオブジェクトで、これはSnowflakeのローダーが最もきれいに読み取る形式です：

# jqを使って
jq -c '.[]' books.raw.json > books.ndjson

# または、jqなしで、Nodeのワンライナー
node -e 'JSON.parse(require("fs").readFileSync("books.raw.json","utf8")).forEach(o=>console.log(JSON.stringify(o)))' > books.ndjson

books.ndjsonには、1行ごとに1つの自己完結型JSONオブジェクトが保持されています。コールドセッションが空のシェルまたは一時的なos error 10054を返す場合は、セッションを閉じ、新しいセッションを作成し、再試行を限定回数行ってから抽出します。

ステップ2 — Snowflakeの準備

ウェアハウス、データベース、スキーマ、JSONファイル形式、および単一のVARIANTカラムを持つ着陸テーブルを作成します。これをsetup.sqlとして保存し、snow sql -c demo -f setup.sqlで実行します：

CREATE WAREHOUSE IF NOT EXISTS ingest_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;
CREATE DATABASE  IF NOT EXISTS web_data;
CREATE SCHEMA    IF NOT EXISTS web_data.raw;

USE WAREHOUSE ingest_wh;
USE SCHEMA web_data.raw;

-- NDJSON: 1行ごとに1つのJSONオブジェクトなので、外部配列を削除しない
CREATE OR REPLACE FILE FORMAT ndjson_format
  TYPE = JSON
  STRIP_OUTER_ARRAY = FALSE
  COMPRESSION = AUTO;

-- 生のレコードをそのまま着陸させる; クエリ時に形を整える
CREATE OR REPLACE TABLE raw_books (
  src       VARIANT,
  loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

STRIP_OUTER_ARRAY = FALSEはNDJSONに対して正しい設定です。なぜなら、各行はすでに独自のオブジェクトだからです — STRIP_OUTER_ARRAY = TRUEは1つの大きな[ ... ]配列のファイルのみに適用されます。

ステップ3 — 方法1: COPY INTOによる一時的なバルクロード

単一ファイルまたは手動バッチの場合、ファイルをステージし、COPY INTOを実行します。最も簡単な方法は、内部名付きステージとPUTを使用することです：

-- JSON形式にバウンドされた名前付き内部ステージ
CREATE OR REPLACE STAGE books_stage FILE_FORMAT = ndjson_format;

# ローカルNDJSONを内部ステージにアップロードします（snow CLIがPUTを実行）
snow sql -c demo -q "PUT file://$(pwd)/books.ndjson @books_stage AUTO_COMPRESS=TRUE OVERWRITE=TRUE"

-- 各オブジェクトをVARIANTカラムに1行としてロードします
COPY INTO raw_books (src)
  FROM @books_stage
  FILE_FORMAT = (FORMAT_NAME = 'ndjson_format')
  ON_ERROR = 'CONTINUE';

JSON キーを VARIANT の代わりに型付きカラムに直接マッピングするには、カラム名がキーと一致するテーブルを作成し、MATCH_BY_COLUMN_NAME を使用します：

CREATE OR REPLACE TABLE books (
  title VARCHAR, price VARCHAR, rating VARCHAR, in_stock BOOLEAN, url VARCHAR
);

COPY INTO books
  FROM @books_stage
  FILE_FORMAT = (TYPE = 'JSON')
  MATCH_BY_COLUMN_NAME = 'CASE_INSENSITIVE';

ステージされたファイルから自動的にスキーマを導出させたい場合は、INFER_SCHEMA と CREATE TABLE … USING TEMPLATE を使用してカラムリストを構築できます：

CREATE OR REPLACE TABLE books_auto
  USING TEMPLATE (
    SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
    FROM TABLE(INFER_SCHEMA(
      LOCATION => '@books_stage',
      FILE_FORMAT => 'ndjson_format'
    ))
  );

すでにクラウドバケットに保存されているデータには、アップロードするのではなく 外部ステージ を指し示します。ストレージ統合を使用します（インラインキーは不要）：

CREATE OR REPLACE STAGE books_s3_stage
  URL = 's3://my-bucket/scraped/books/'
  STORAGE_INTEGRATION = my_s3_integration
  FILE_FORMAT = ndjson_format;

COPY INTO raw_books (src) FROM @books_s3_stage;

無料プランで API キーを取得する： app.scrapeless.com

ステップ 4 — メソッド 2: Snowpipe を使用した継続的バッチ

スクレーパーがスケジュールに従って新しいファイルをバケットに落とすと、Snowpipe が各ファイルを自動的にロードします — 手動の COPY や専用のウェアハウスは不要です。パイプは COPY INTO ステートメントをラップします；AUTO_INGEST = TRUE にすると、クラウドイベント通知がロードをトリガーします：

CREATE OR REPLACE PIPE books_pipe
  AUTO_INGEST = TRUE
  AWS_SNS_TOPIC = 'arn:aws:sns:us-east-1:123456789012:scraped-bucket'
  AS
  COPY INTO raw_books (src)
  FROM @books_s3_stage
  FILE_FORMAT = (TYPE = 'JSON');

S3 ではイベントは SNS/SQS を介して Snowflake 管理のキューに流れます；GCS は Pub/Sub、Azure は Event Grid を使用し、それぞれ通知統合が接続されています。Snowpipe を明示的に呼び出したい場合は、AUTO_INGEST を設定せずにステージされたファイルパスを insertFiles REST エンドポイントに POST し、その後 insertReport をポーリングします。

Snowflake のガイダンスからの二つの運用ノート：

• 請求はサーバーレス で、Snowpipe が取り込むデータについて GB ごとに請求されます — サイジングするウェアハウスはなく、以前のファイルごとコンポーネントは廃止されました。
• ファイルサイズが重要です。 圧縮されたファイルを 100–250 MB 程度に設定し、1 分あたり 1 回程度の頻度でステージします；それ以上の頻度でステージすると、レイテンシを減らすことなくキュー管理のオーバーヘッドが追加されます。小さなスクレーピングバッチを大きなファイルにバッファリングしてからステージします。

Snowpipe はデータを数分以内で利用可能にし、それは予定された市場スナップショットに適しています。

ステップ 5 — メソッド 3: Snowpipe Streaming を使用した低レイテンシ行

新鮮さが秒単位で必要な場合、Snowpipe Streaming は行をテーブルに直接書き込みます — ステージされたファイルは不要です。高性能アーキテクチャは 2025 年 9 月から一般提供されています で、Java、Python、Node.js の SDK と共有クライアントコアに対する REST API が用意されています；クラシックなファイルベースの Snowpipe Streaming は非推奨の道を歩んでいます。

モデルには三つのコア概念があります：

• チャネル — テーブルへの名前付き、長期的なストリーミング接続。行はチャネル内で順序通りにコミットされます。
• オフセットトークン — アプリケーションが各バッチに添付する文字列。再起動後、getLatestCommittedOffsetToken() によって最後に耐久的にコミットされた位置を知ることができるため、次に続くものだけをリプレイできます — 正確な一回配信の基礎です。
• スループット請求 — ファイルごとではなく、取り込まれた未圧縮 GB あたりのクレジットです。

Java クライアントの形状は小さいです：

SnowflakeStreamingIngestClient client =
    SnowflakeStreamingIngestClientFactory.builder("BOOKS_CLIENT")
        .setProperties(props).build();

OpenChannelRequest request = OpenChannelRequest.builder("BOOKS_CHANNEL")
    .setDBName("WEB_DATA").setSchemaName("RAW").setTableName("RAW_BOOKS")
    .setOnErrorOption(OpenChannelRequest.OnErrorOption.CONTINUE)
    .build();

SnowflakeStreamingIngestChannel channel = client.openChannel(request);
channel.insertRow(rowAsMap, offsetToken);   // 一つのスクレープされたレコード
channel.getLatestCommittedOffsetToken();     // リカバリ用

スクレープされたレコードが継続的に到着し、ダッシュボードが数秒以内にそれらを必要とする場合は、Streaming を選択します。

ステップ 6 — メソッド 4: Kafka コネクタを使用したイベント駆動のロード

もしスクレープされたレコードがすでに Apache Kafka 上を流れているなら、Snowflake Connector for Kafka はトピックをテーブルにロードします（1 つのトピックが 1 つのテーブルにマッピングされます）。これを Kafka Connect ワーカーの内部で実行します。単一のプロパティがその下の取り込みエンジンを選択します：

name=scraped-books-sink
connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
topics=scraped_books
snowflake.database.name=WEB_DATA
snowflake.schema.name=RAW
# SNOWPIPE （ファイルベース、デフォルト）または SNOWPIPE_STREAMING （低レイテンシ）
snowflake.ingestion.method=SNOWPIPE_STREAMING

各テーブルに接続子が作成する2つの VARIANT 列があります： RECORD_CONTENT（メッセージペイロード）と RECORD_METADATA（トピック、パーティション、オフセット、タイムスタンプ、ヘッダー）。ペイロードは、任意の VARIANT と同じようにクエリを実行します。

完全に管理された代替案 — Kafka Connect クラスターを操作する必要がない — は、Snowflake Openflow（一般的に利用可能で、Apache NiFi に基づいて構築されています）が、Kafka、Kinesis、データベース、および SaaS ソースから Snowflake へ管理されたパイプラインを通じてデータを取り込みます。

手法の選択

ほとんどのチームはスキーマを検証するために COPY INTO から始め、スクレイパーがスケジュールに従って実行されるようになったら Snowpipe に移行し、サブミニットのフレッシュさが追加の移動部分を正当化する場合にのみ Streaming または Kafka を採用します。

読み込まれたデータのクエリ

生のレコードは VARIANT に存在するため、読み込み時にフォーマットします。 : 演算子を使ってナビゲートし、 :: でキャストします：

SELECT
  src:title::string   AS title,
  src:price::string   AS price,
  src:rating::string  AS rating,
  src:in_stock::boolean AS in_stock,
  src:url::string     AS url
FROM raw_books;

スクレイピングされたレコードが配列を持つ場合 — photos リストや priceHistory シリーズ — LATERAL FLATTEN で行に展開します：

-- raw_listings: raw_books と同様の方法でロードされた仮想のスクレイプされたリスティングテーブル。
-- （書籍の例にはネストされた配列がありません；これはそれを持つソースのパターンを示しています。）
SELECT
  src:title::string AS title,
  ph.value:date::string  AS price_date,
  ph.value:price::number AS price
FROM raw_listings,
LATERAL FLATTEN(INPUT => src:priceHistory) ph;

ソースページがフィールドを追加してもマイグレーションは必要ありません — 次のロードでは src に単に表示されます。

返される内容

COPY INTO の後、各スクレイプされたオブジェクトは raw_books の1行になります。 VARIANT 列にはレコードがそのまま保持され、以下のスキーマは標準的であり、フィールド値は例として示されています：

// 1行の src VARIANT, ステップ1の抽出器によって生成されました。
{
  "title": "A Light in the Attic",
  "price": "£51.77",
  "rating": "Three",
  "in_stock": true,
  "url": "https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html"
}

いくつかの正直な観察：

• 価格は表示文字列として到着します。 ページから値を期待するのではなく、SQL でキャストしてクリーンにします（ REPLACE(src:price::string, '£', '')::number）。
• 条件付きフィールドは Nullable です。 特定のページに欠落しているフィールドは単に src からMissingで、 VARIANT アクセスは NULL を返します。
• MATCH_BY_COLUMN_NAME は一致しないキーをスキップします。 新しいキーは自動的に VARIANT テーブルに配置されますが、型付きテーブルは列を追加するまでそれらを破棄します。
• Snowpipe のロード履歴は14日間保持されます パイプメタデータに；一括の COPY 履歴は64日間テーブルメタデータに保持されます — バックフィルを監査する際にはそれを考慮してください。

結論

スクレイピングされたウェブデータを Snowflake に取り込むには、4つの手順が必要です：Scrapeless でレンダリングと抽出を行い、NDJSON を生成し、ファイルをステージし、レイテンシに適した手法でロードします — COPY INTO、Snowpipe、Snowpipe Streaming、または Kafka コネクタ。このパイプラインは、ソースサイトがフィールドを追加しても生き残るように、 VARIANT 列に生のレコードを格納し、クエリ時にフォーマットします。

対象が必要なときに米国の出口をピン留めし、Scrapeless セッションチェーンを1つのシェルの呼び出し内に保持し、発見 → 抽出パターンに従い、欠落しているフィールドを Nullable として扱います。ここで使われているサンドボックスよりも難しいターゲットには、同じプロデューサーパターンが適用されます — 2026 年のベスト Zillow スクレイパー ガイド、Scraping Browser 製品ページ、および Scrapeless ドキュメントを参照してください。

AI駆動のデータパイプラインの構築準備はできましたか？

無料プランを請求し、ウェブデータからデータウェアハウスへのパイプラインを構築している開発者とつながるために、コミュニティに参加しましょう：Discord · Telegram。

Scrapeless Webサイトにサインアップして、無料のScraping Browser実行環境を入手し、scrapeless.com/en/pricingを見て、パイプラインの成長に合わせてセッション時間と同時実行をスケールアップしてください。

よくある質問

Q1：データウェアハウスへの取り込みのためのウェブスクレイピングは合法ですか？
公に見えるデータを収集することは一般的に正当化されますが、合法性はターゲットサイトの利用規約、法的管轄、およびデータの種類によって異なります。サイトの利用規約を確認し、個人情報や制限されたデータを避け、商業利用の前に法律の専門家に相談してください。ここで使用されているサンドボックスサイト books.toscrape.com は、スクレイピングの練習のために特に存在しています。

Q2: NDJSON か JSON 配列、どちらをスクレイパーが出力すべきですか？

NDJSON（一行ごとにオブジェクト）は最もクリーンに読み込まれ、ファイル全体をバッファリングせずにストリーミングされます。STRIP_OUTER_ARRAY = FALSE を設定してください。プロデューサーが単一の [ ... ] 配列を出力する場合は、STRIP_OUTER_ARRAY = TRUE に設定し、各要素を行として扱います。

Q3: VARIANT 列にロードするべきか、それとも型付き列にするべきか？

生のスクレイピングデータを VARIANT に保存し、SQL で整形します。ソースページは変わるため、VARIANT は新しいフィールドを移行なしで吸収します。スキーマが安定するまでは、型付き列には MATCH_BY_COLUMN_NAME を一度だけ使用してください。

Q4: どの取り込み方法を選ぶべきですか？

一回限りのロードには COPY INTO を使用し、スケジュールされたファイルバッチには Snowpipe（数分の遅延、サーバーレス）を、サブミニットの行レベルの新鮮さには Snowpipe Streaming を、Kafka 経由でレコードが既に流れている場合は Kafka コネクタを使用します。まずは COPY INTO から始め、必要に応じて新鮮さが必要な場合は進んでください。

Q5: os error 10054 や 503 のような一時的なスクレイパーエラーをどう処理すればよいですか？

それらは一時的なものと見なします：Scrapeless セッションを閉じ、新たなセッションを作成し、再度ナビゲートし、抽出前に安定したセレクターを待ってください。再試行は制限を設けてください。これらはプロデューサー側に属し、Snowflake には影響を与えず、ステージにあるものをすべてロードします。

Q6: Snowpipe に対してウェアハウスを稼働させる必要がありますか？

いいえ。Snowpipe はサーバーレスで、取り込まれた GB あたり請求されます — Snowflake が計算リソースを提供します。ユーザー管理のウェアハウスは、バルク COPY INTO とクエリ処理のためにのみ必要です。

Q7: AI エージェントなしでこれを実行できますか？

はい。scrapeless-scraping-browser CLI は、プレーンなシェルからエンドツーエンドで NDJSON を生成し、Snowflake 側は普通の SQL です。MCP 接続されたエージェントは便利な道ですが、必須ではありません。