執筆日：2026年5月15日

免責事項：本記事は執筆時点の情報に基づいています。内容は将来変更される可能性があります。

前提条件

Snowflakeの基本操作が可能であること
Snowsightのアクセス権限を有していること

はじめに

はじめまして。本ブログを閲覧いただきありがとうございます。株式会社NTTデータ九州ビジネス共創部デジタルビジネス推進室D&I（Data & Intelligence）チームの佐藤です。

D&Iチームでは、TableauやAlteryx、SnowflakeなどのD&I（データ＆インテリジェンス）領域におけるツール導入支援や開発を通じて、お客様の業務／意思決定の効率化・高度化のご支援に取り組んでいます。

今回は「非構造データの取込（構造化）」をテーマに、SnowparkとStreamlit（Streamlit in Snowflake）を使用した取込処理の仕組化（アプリケーションとして実行するところまで）を検証してみました。

実際にe-Statから取得したデータを対象に検証していますので、是非とも本記事を参考にお試しいただけますと幸いです。
※ソースコードについては、手動実行ができるところまでを記事に掲載しております。アプリケーション作成時のソースコードについては掲載しておりませんので、お試しいただく際は予めご容赦ください。

※Snowparkの詳しい機能につきましては、以下公式サイトをご参照ください。

Snowpark：Python、Java、Scalaなど、任意の言語での構築を実現

※Streamlit in Snowflakeの詳しい機能につきましては、以下公式サイトをご参照ください。

Streamlit in Snowflake について | Snowflake Documentation

※ 本記事では、e-Stat（日本の統計が閲覧できる政府統計ポータルサイト）にて取得したオープンデータを利用しています。

政府統計の総合窓口

背景・実現したい仕組みの全体像

まず、そもそもの話になるのですが、
この検証をやってみようと思った背景としては、以下のような悩みを持つ方が多いのでは？と思ったことがきっかけとなります。

自社でいつも作成しているExcelファイル（非構造）をきちんとデータ化して、いつでも分析できるようにしたい
データ化を人力で行っているが、毎回時間がかかってしまう。また、人力でやっているのでヒューマンエラー等によりいつも修正対応に追われている。
どのファイルがデータ化が完了していて、どのファイルが完了していないのかを追えていない。
最初から構造化されたデータを業務に用いることも検討したが、従来の様式での報告（合意形成、意思決定など）も必要となるので、できればそのままにしておきたい。

これらの悩みを解消できるような簡易的な仕組みを構築し、以下のようなオペレーションで簡単にデータ化できるようなものを作っていこうと思います。

非構造データを取得後、Snowflake上のステージにファイルをアップロード
アプリケーション（Streamlitで作成）で未処理ファイルを選択
同アプリケーション内でデータ化処理実行（Snowparkを活用）
同アプリケーション内で結果確認　→　データ化完了！

実現したい仕組みの全体像（イメージ）は以下となります。
※生成AIによるイメージ生成となりますので、何卒ご容赦ください。

検証の流れ

大まかな検証の流れは以下の通りです。

主に、Snowsightを利用して実施します。

事前準備（データ）
e-Statからデータをダウンロードし、ファイル名称を変更
事前準備（環境）
検証用の各種オブジェクトを作成し、ステージにファイルを手動アップロード
処理部分作成
プロシージャを作成し、取込テンプレを登録
手動実行確認
手動でプロシージャを実行し、データ取込が正常に実施できることを確認
アプリケーション作成
GUIで操作できるような運用コンソールを作成
触ってみる
作成したアプリケーションを使用して実際にデータ取込を実行してみる

なお、今回のSnowflake環境では、エディションにビジネスクリティカル、プラットフォームにはAmazon Web Servicesの東京リージョンを採用しています。

上記以外の環境の場合、手順が異なる場合がありますのでご留意ください。

1. 事前準備（データ）

サンプルとして使用するデータを取得します。

今回はe-Statから「労働力調査 / 就業状態別15歳人口、産業別就業者数、完全失業者数（政府統計コード：00200531）」の月次データ（最初は1ファイルのみを対象としますが、最後に複数年月分をアップロードします）をダウンロードします。

※以下リンクは2025年1月のものとなります。

労働力調査労働力調査（公表資料、時系列結果など）結果の概要（各期の主な結果をまとめた資料・報道資料）基本集計最新月分結果 1 就業状態別15歳以上人口，産業別就業者数，完全失業者数月次 2025年1月 | ファイル | 統計データを探す | 政府統計の総合窓口

複数年月分のダウンロードは、以下リンク先で年月を指定しそれぞれダウンロードください。

労働力調査労働力調査（公表資料、時系列結果など）結果の概要（各期の主な結果をまとめた資料・報道資料）基本集計 | ファイル | 統計データを探す | 政府統計の総合窓口

表示・ダウンロードボタンを押下します。

ダウンロードしたExcelファイルを以下の命名規則に則ってリネームします。

・労働力調査_月次_yyyymm.xlsx

一気に複数ファイルをダウンロードした方は、すべてリネームしておきましょう。

また、今回取得したデータについて、軽く中身を見ておきます。

特徴としては、

表形式ではあるが、表のタイトルや年月情報がシート上部に記載されている。それに伴い、A1セルでないところに表が存在する。
セル結合が多用されている。
データ化を行う際に、ディメンション候補となる分類項目（例えば、[人口]や[雇就業状態]など）が階層構造になっている。
メジャー項目についても、男女の区分や実数、対前年同月増減や増減率に分かれて整理されている。

となります。

本来、こういったデータをデータ分析等で「利用できる形」にするのは一苦労かかりますが、今回はこのデータを対象に綺麗に取り込んでいきたいと思います。

2. 事前準備（環境）

次に、Snowflake側（Snowsight）で作業していきます。

Snowsightにアクセスし、ナビゲーションバーの「プロジェクト」→「ワークスペース」に移動してください。

任意のワークスペース上で「＋新規追加」を押下し、SQLファイルを作成します。

作成したSQLファイル内に下記コードをコピーし、上から順に実行していきます。

やっていることとしては、

検証用ロールの作成と自ユーザへのロール付与
ウェアハウス、データベース、スキーマ、ステージの作成
メタテーブルの作成（取込対象となるファイルを原本管理する、そして取込時のログを保管するためのメタテーブルを作成）
取込テンプレテーブルの作成（ファイル形式毎の取込設定を管理するためのテーブルを作成）
出力テーブルと提供ビューの作成（構造化処理後に、縦持ちデータを出力テーブルに格納し、横持ちデータを提供ビューとして保管）

となります。

----------------------------------------------- 1. 検証用ロール作成

-- 検証用ロールを作成し権限を設定
USE ROLE ACCOUNTADMIN;
CREATE ROLE IF NOT EXISTS "TEST_ROLE";
GRANT ROLE SYSADMIN TO ROLE "TEST_ROLE";

-- 自分に付与（今回は自身 TEST ユーザに付与、必要に応じて変更してください）
GRANT ROLE "TEST_ROLE" TO USER TEST;

----------------------------------------------- 2. ウェアハウス、データベース、スキーマ、ステージ作成

USE ROLE "TEST_ROLE";

-- ウェアハウス作成
CREATE WAREHOUSE IF NOT EXISTS SAMPLE_WH
WAREHOUSE_SIZE = 'XSMALL'
AUTO_SUSPEND = 60
AUTO_RESUME = TRUE
INITIALLY_SUSPENDED = TRUE;

-- データベース作成
CREATE DATABASE IF NOT EXISTS "SAMPLE_DB";

-- スキーマ作成
CREATE SCHEMA IF NOT EXISTS "SAMPLE_DB"."SAMPLE_SCHEMA";

-- ステージ作成
USE ROLE "TEST_ROLE";
USE WAREHOUSE SAMPLE_WH;
USE DATABASE "SAMPLE_DB";
USE SCHEMA "SAMPLE_SCHEMA";
CREATE STAGE IF NOT EXISTS "取込ステージ" DIRECTORY = (ENABLE = TRUE);


----------------------------------------------- 3. メタテーブル作成（3つ）

USE ROLE "TEST_ROLE";
USE WAREHOUSE SAMPLE_WH;
USE DATABASE "SAMPLE_DB";
USE SCHEMA "SAMPLE_SCHEMA";

-- メタテーブル（原本ファイル管理）作成
CREATE TABLE IF NOT EXISTS "原本ファイル管理" (
"原本ファイルID" NUMBER AUTOINCREMENT,
"ステージパス" STRING,
"ファイル名" STRING,
"拡張子" STRING,
"登録日時" TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
"ステータス" STRING DEFAULT '未処理',
"最新実行ID" NUMBER,
"備考" STRING,
PRIMARY KEY ("原本ファイルID")
);

-- メタテーブル（取込実行ログ）作成
CREATE TABLE IF NOT EXISTS "取込実行ログ" (
"実行ID" NUMBER AUTOINCREMENT,
"原本ファイルID" NUMBER,
"開始日時" TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
"終了日時" TIMESTAMP_NTZ,
"実行結果" STRING,
"読込行数" NUMBER,
"出力行数" NUMBER,
"適用テンプレID" STRING,
"実行者" STRING DEFAULT CURRENT_USER(),
"メッセージ" STRING,
PRIMARY KEY ("実行ID")
);

-- メタテーブル（取込エラーログ）作成
CREATE TABLE IF NOT EXISTS "取込エラーログ" (
"エラーID" NUMBER AUTOINCREMENT,
"実行ID" NUMBER,
"原本ファイルID" NUMBER,
"エラー分類" STRING,
"エラーメッセージ" STRING,
"エラー詳細" VARIANT,
"発生日時" TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
PRIMARY KEY ("エラーID")
);

----------------------------------------------- 4. 取込テンプレテーブル作成

USE ROLE "TEST_ROLE";
USE WAREHOUSE SAMPLE_WH;
USE DATABASE "SAMPLE_DB";
USE SCHEMA "SAMPLE_SCHEMA";

-- 取込テンプレテーブル作成
CREATE TABLE IF NOT EXISTS "取込テンプレ" (
"テンプレID" STRING,
"テンプレ名" STRING,
"適用条件_ファイル名正規表現" STRING,
"命名規則_説明" STRING,
"命名規則_正規表現" STRING,
"実行SP名" STRING,
"設定JSON" VARIANT,
"有効フラグ" BOOLEAN DEFAULT TRUE,
"作成日時" TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
"更新日時" TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

----------------------------------------------- 5. 出力テーブル（縦持ち）と提供ビュー（横持ち）作成

USE ROLE "TEST_ROLE";
USE WAREHOUSE SAMPLE_WH;
USE DATABASE "SAMPLE_DB";
USE SCHEMA "SAMPLE_SCHEMA";

-- 出力テーブル作成
CREATE OR REPLACE TABLE "TBL_労働力調査_月次積上" (
  "構造化データID"  NUMBER AUTOINCREMENT,
  "原本ファイルID"  NUMBER,
  "年月"            STRING,
  "分類1"           STRING,
  "分類2"           STRING,
  "分類3"           STRING,
  "分類4"           STRING,
  "分類5"           STRING,
  "分類6"           STRING,
  "男女区分"        STRING,
  "指標"            STRING,
  "値"              FLOAT,
  "単位"            STRING,
  "注釈"            STRING,
  "取込日時"        TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  "実行ID"          NUMBER,
  PRIMARY KEY ("構造化データID")
);

-- 提供ビュー作成
CREATE OR REPLACE VIEW "VIE_労働力調査_月次積上" AS
SELECT
  "原本ファイルID",
  "年月",
  "分類1","分類2","分類3","分類4","分類5","分類6",
  "男女区分",
  MAX(CASE WHEN "指標"='実数' THEN "値" END) AS "実数",
  MAX(CASE WHEN "指標"='対前年同月-増減' THEN "値" END) AS "対前年同月-増減",
  MAX(CASE WHEN "指標"='対前年同月-増減率(%)' THEN "値" END) AS "対前年同月-増減率(%)",
  MAX("取込日時") AS "取込日時",
  MAX("実行ID") AS "実行ID"
FROM "TBL_労働力調査_月次積上"
GROUP BY
  "原本ファイルID","年月",
  "分類1","分類2","分類3","分類4","分類5","分類6",
  "男女区分";

すべて実行完了後、各オブジェクトが作成されていることを確認してください。

また、作成したステージ（取込ステージ）にダウンロードしたExcelを手動でアップロードしておきます。

データベースエクスプローラー上で「取込ステージ」を選択し、「＋ファイル」を押下します。

「ここにファイルをドラッグアンドドロップ」と記載されているエリアに該当ファイルをドラッグアンドドロップし、「アップロード」を押下します。

ステージファイルとして無事に追加されたら事前準備は完了です。

3. 処理部分作成

実際にデータを構造化処理するための処理部分（プロシージャ）を作成していきます。

ここで活躍するのがSnowparkになります。

Snowparkを簡単に説明すると、

Snowflakeが提供する開発フレームワーク
SQL以外の開発言語（今回はPython）のコードを実行することが可能

といった感じです。

事前準備の時と同じく、ワークスペースにて新しいSQLファイルを新規追加し、
下記コードを上から順に実行していきます。

やっていることとしては、

プロシージャ（SP）の作成
取込テンプレの登録

となります。

※下記コードは Snowflake Cortex Code（AIコーディングエージェント）等の生成AIを活用し作成したものとなりますので、何卒ご容赦ください。

※下記コードではopenpyxl 等のOSSライブラリを利用しています。利用にあたっては各ライブラリのライセンス条項をご確認ください。

Cortex Code：SnowflakeネイティブなAIデータコーディングエージェント

----------------------------------------------- 1. プロシージャ作成（Snowpark）

USE ROLE "TEST_ROLE";
USE WAREHOUSE SAMPLE_WH;
USE DATABASE "SAMPLE_DB";
USE SCHEMA "SAMPLE_SCHEMA";

CREATE OR REPLACE PROCEDURE "SAMPLE_DB"."SAMPLE_SCHEMA"."SP_労働力調査_月次積上_取込"("原本ファイルID" NUMBER)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.10'
PACKAGES = ('snowflake-snowpark-python','openpyxl')
HANDLER = 'run'
AS
$$
import re
import json
from snowflake.snowpark.files import SnowflakeFile
import openpyxl

DB_NAME = "SAMPLE_DB"
SCHEMA_NAME = "SAMPLE_SCHEMA"

DBQ = f'"{DB_NAME}"'
SCHEMAQ = f'"{SCHEMA_NAME}"'

T_RAW = f'{DBQ}.{SCHEMAQ}."原本ファイル管理"'
T_RUN = f'{DBQ}.{SCHEMAQ}."取込実行ログ"'
T_ERR = f'{DBQ}.{SCHEMAQ}."取込エラーログ"'
T_TPL = f'{DBQ}.{SCHEMAQ}."取込テンプレ"'
T_OUT = f'{DBQ}.{SCHEMAQ}."TBL_労働力調査_月次積上"'
T_WORK = f'{DBQ}.{SCHEMAQ}."WK_労働力調査_月次積上_挿入"'

_ZEN2HAN = str.maketrans("０１２３４５６７８９", "0123456789")

def _q(x):
    if x is None:
        return ""
    return str(x).replace("'", "''")

def _norm_text_keep_brackets(x):
    if x is None:
        return None
    s = str(x).translate(_ZEN2HAN)
    s = s.replace("\u3000", " ")
    s = s.strip()
    s = re.sub(r"[ \t]+", "", s)
    return s if s != "" else None

def _cell_value(ws, r, c):
    v = ws.cell(r, c).value
    if v is not None:
        return v
    try:
        for rg in ws.merged_cells.ranges:
            if rg.min_row <= r <= rg.max_row and rg.min_col <= c <= rg.max_col:
                return ws.cell(rg.min_row, rg.min_col).value
    except Exception:
        pass
    return None

def _normalize_regex_pattern(pat_text):
    if pat_text is None:
        return pat_text
    s = str(pat_text)
    def looks_overescaped(x):
        return ("\\\\d" in x) or ("\\\\s" in x) or ("\\\\w" in x) or ("\\\\b" in x)
    for _ in range(3):
        if looks_overescaped(s):
            s = s.replace("\\\\", "\\")
        else:
            break
    return s

def _extract_ym(ws, ym_conf):
    ym_r = int(ym_conf.get("行", 3))
    ym_c = int(ym_conf.get("列", 2))
    raw_pat = ym_conf.get("正規表現")
    if raw_pat is None or str(raw_pat).strip() == "":
        raw_pat = r"([0-9]{4})\s*年\s*([0-9]{1,2})\s*月"
    pat_text = _normalize_regex_pattern(raw_pat)
    ym_pat = re.compile(pat_text)

    def try_parse(v):
        if v is None:
            return None
        s = str(v).translate(_ZEN2HAN)
        m = ym_pat.search(s)
        if m:
            return f"{m.group(1)}-{int(m.group(2)):02d}"
        s2 = re.sub(r"[ \t\u3000]+", "", s)
        m = ym_pat.search(s2)
        if m:
            return f"{m.group(1)}-{int(m.group(2)):02d}"
        mY = re.search(r"([0-9]{4})\s*年", s)
        mM = re.search(r"([0-9]{1,2})\s*月", s)
        if mY and mM:
            return f"{mY.group(1)}-{int(mM.group(1)):02d}"
        return None

    v0 = _cell_value(ws, ym_r, ym_c)
    ym = try_parse(v0)
    if ym:
        return ym
    parts = []
    for cc in range(ym_c, min(ym_c + 8, ws.max_column) + 1):
        vv = _cell_value(ws, ym_r, cc)
        if vv is not None:
            parts.append(str(vv))
    ym = try_parse("".join(parts))
    if ym:
        return ym
    max_r2 = min(ws.max_row, 60)
    max_c2 = min(ws.max_column, 40)
    for rr in range(1, max_r2 + 1):
        for cc in range(1, max_c2 + 1):
            ym = try_parse(_cell_value(ws, rr, cc))
            if ym:
                return ym
    raise Exception("年月が抽出できません")

def run(session, fid: int):
    r = session.sql(f"""
        SELECT "ステージパス","ファイル名"
        FROM {T_RAW}
        WHERE "原本ファイルID" = {fid}
    """).collect()
    if not r:
        return {"結果":"失敗","理由":"原本ファイル管理に該当IDなし","原本ファイルID":fid}

    stage_path = r[0]["ステージパス"]
    file_name = r[0]["ファイル名"]

    session.sql(f"""
        INSERT INTO {T_RUN} ("原本ファイルID","開始日時","実行結果","メッセージ")
        VALUES ({fid}, CURRENT_TIMESTAMP(), '処理中', 'Snowpark取込開始')
    """).collect()

    run_id = session.sql(f"""
        SELECT MAX("実行ID") AS "実行ID"
        FROM {T_RUN}
        WHERE "原本ファイルID" = {fid}
    """).collect()[0]["実行ID"]

    session.sql(f"""
        UPDATE {T_RAW}
        SET "ステータス"='処理中', "最新実行ID"={run_id}
        WHERE "原本ファイルID"={fid}
    """).collect()

    try:
        tmpl = session.sql(f"""
            SELECT "テンプレID","設定JSON"
            FROM {T_TPL}
            WHERE "有効フラグ"=TRUE
              AND REGEXP_LIKE('{_q(file_name)}', "適用条件_ファイル名正規表現")
            ORDER BY "更新日時" DESC
            LIMIT 1
        """).collect()
        if not tmpl:
            raise Exception(f"テンプレ未検出（ファイル名={file_name}）")

        tpl_id = tmpl[0]["テンプレID"]
        conf = tmpl[0]["設定JSON"]
        if isinstance(conf, str):
            conf = json.loads(conf)

        sheet_name = conf.get("シート名", "第1表")

        m2 = re.match(r'^@([^/]+)/(.+)$', str(stage_path))
        if not m2:
            raise Exception(f"ステージパス形式不正: {stage_path}")
        stage_only = m2.group(1)
        rel_path = m2.group(2)

        stage_ref_fq = f'@{DBQ}.{SCHEMAQ}."{stage_only}"'
        scoped_url = session.sql(
            f"SELECT BUILD_SCOPED_FILE_URL('{_q(stage_ref_fq)}','{_q(rel_path)}') AS URL"
        ).collect()[0]["URL"]

        with SnowflakeFile.open(scoped_url, 'rb') as f:
            wb = openpyxl.load_workbook(f, data_only=True)

        if sheet_name not in wb.sheetnames:
            raise Exception(f"シート名が見つかりません: {sheet_name}")
        ws = wb[sheet_name]

        ym = _extract_ym(ws, conf.get("年月セル候補", {}))

        unit_conf = conf.get("単位セル候補", {})
        u_r, u_c = int(unit_conf.get("行", 3)), int(unit_conf.get("列", 15))
        unit_text = _cell_value(ws, u_r, u_c)
        unit_text = "" if unit_text is None else str(unit_text)
        for x in unit_conf.get("前後除去", []):
            unit_text = unit_text.replace(x, "")
        unit_val = unit_text.strip()
        unit_val = unit_val if unit_val != "" else None

        start_row = int(conf.get("データ開始行", 10))
        end_row = int(conf.get("データ終了行", ws.max_row))
        num_start_col = int(conf.get("数値開始列", 7))

        gender_map = conf.get("男女区分マッピング", [])
        metric_map = conf.get("指標マッピング", [])

        cat1 = None
        cat2 = None
        lv = [None, None, None, None]

        big_re = re.compile(r"^－.*－$")
        br_re = re.compile(r"^〔.*〕$")

        inserts = []
        read_cnt = 0

        max_r = min(ws.max_row, end_row)
        for rno in range(1, max_r + 1):
            a = _norm_text_keep_brackets(_cell_value(ws, rno, 1))
            b = _norm_text_keep_brackets(ws.cell(rno, 2).value)
            c = _norm_text_keep_brackets(ws.cell(rno, 3).value)
            d = _norm_text_keep_brackets(ws.cell(rno, 4).value)
            e = _norm_text_keep_brackets(ws.cell(rno, 5).value)

            if a and big_re.match(a):
                if "原数値" in a:
                    cat1 = "原数値"
                elif "季節調整値" in a:
                    cat1 = "季節調整値"
                else:
                    cat1 = a.replace("－", "")
                cat2 = None
                lv = [None, None, None, None]
                continue

            if a and br_re.match(a):
                cat2 = _norm_text_keep_brackets(a.replace("〔", "").replace("〕", ""))
                lv = [None, None, None, None]
                continue

            if rno < start_row:
                continue

            mapped = [b, c, d, e]
            updated_idx = -1
            for i in range(4):
                if mapped[i] is not None:
                    lv[i] = mapped[i]
                    updated_idx = i
            if updated_idx >= 0:
                for i in range(updated_idx + 1, 4):
                    if mapped[i] is None:
                        lv[i] = None

            any_num = False
            for cc in range(num_start_col, num_start_col + 9):
                v = _cell_value(ws, rno, cc)
                if isinstance(v, (int, float)):
                    any_num = True
                    break
            if not any_num:
                continue

            read_cnt += 1

            for g in gender_map:
                gender = g.get("男女区分")
                base_col = int(g.get("開始列"))
                for mm in metric_map:
                    metric = mm.get("指標")
                    off = int(mm.get("列オフセット"))
                    val = _cell_value(ws, rno, base_col + off)
                    if val is None:
                        continue
                    try:
                        val_f = float(val)
                    except:
                        continue

                    inserts.append((
                        fid,
                        ym,
                        cat1,
                        cat2,
                        lv[0], lv[1], lv[2], lv[3],
                        gender,
                        metric,
                        val_f,
                        unit_val,
                        None,
                        run_id
                    ))

        out_cnt = len(inserts)

        if inserts:
            df = session.create_dataframe(
                inserts,
                schema=[
                    "原本ファイルID","年月",
                    "分類1","分類2","分類3","分類4","分類5","分類6",
                    "男女区分","指標","値","単位","注釈","実行ID"
                ]
            )
            df.write.mode("overwrite").save_as_table(T_WORK)
            session.sql(f"""
                INSERT INTO {T_OUT}
                ("原本ファイルID","年月",
                 "分類1","分類2","分類3","分類4","分類5","分類6",
                 "男女区分","指標","値","単位","注釈","実行ID")
                SELECT
                 "原本ファイルID","年月",
                 "分類1","分類2","分類3","分類4","分類5","分類6",
                 "男女区分","指標","値","単位","注釈","実行ID"
                FROM {T_WORK}
            """).collect()
            session.sql(f'DROP TABLE IF EXISTS {T_WORK}').collect()

        session.sql(f"""
            UPDATE {T_RUN}
            SET "終了日時"=CURRENT_TIMESTAMP(),
                "実行結果"='成功',
                "読込行数"={read_cnt},
                "出力行数"={out_cnt},
                "適用テンプレID"='{_q(tpl_id)}',
                "メッセージ"='Snowpark取込成功'
            WHERE "実行ID"={run_id}
        """).collect()

        session.sql(f"""
            UPDATE {T_RAW}
            SET "ステータス"='成功', "最新実行ID"={run_id}
            WHERE "原本ファイルID"={fid}
        """).collect()

        return {
            "結果":"成功",
            "原本ファイルID":fid,
            "実行ID":run_id,
            "テンプレID":tpl_id,
            "年月":ym,
            "読込行数":read_cnt,
            "出力行数":out_cnt
        }

    except Exception as ex:
        err = str(ex)
        session.sql(f"""
            INSERT INTO {T_ERR}
              ("実行ID","原本ファイルID","エラー分類","エラーメッセージ","エラー詳細")
            SELECT
              {run_id},
              {fid},
              '処理失敗',
              '{_q(err)}',
              OBJECT_CONSTRUCT('例外','{_q(err)}')
        """).collect()
        session.sql(f"""
            UPDATE {T_RUN}
            SET "終了日時"=CURRENT_TIMESTAMP(),
                "実行結果"='失敗',
                "メッセージ"='{_q(err)}'
            WHERE "実行ID"={run_id}
        """).collect()
        session.sql(f"""
            UPDATE {T_RAW}
            SET "ステータス"='失敗', "最新実行ID"={run_id}
            WHERE "原本ファイルID"={fid}
        """).collect()
        return {"結果":"失敗","原本ファイルID":fid,"実行ID":run_id,"理由":err}
$$;

----------------------------------------------- 2. 取込テンプレ登録

MERGE INTO "取込テンプレ" t
USING (
  SELECT
    'T_05401' AS "テンプレID",
    '就業状態別（05401）' AS "テンプレ名",
    '^労働力調査_月次_\\d{6}\\.xlsx$' AS "適用条件_ファイル名正規表現",
    '労働力調査_月次_YYYYMM.xlsx（例：労働力調査_月次_202401.xlsx）' AS "命名規則_説明",
    '^労働力調査_月次_\\d{6}\\.xlsx$' AS "命名規則_正規表現",
    '"SP_労働力調査_月次積上_取込"' AS "実行SP名",
    PARSE_JSON($$
    {
      "シート名": "第1表",

      "年月セル候補": { "行": 3, "列": 2, "正規表現": "(\\\\d{4})年\\\\s*(\\\\d{1,2})月" },
      "単位セル候補": { "行": 3, "列": 15, "前後除去": ["（", "）", "(", ")", "　", " "] },

      "データ開始行": 10,
      "データ終了行": 500,

      "数値開始列": 7,

      "分類列": [2,3,4,5,6],

      "男女区分マッピング": [
        { "男女区分": "男女計", "開始列": 7 },
        { "男女区分": "男",     "開始列": 10 },
        { "男女区分": "女",     "開始列": 13 }
      ],

      "指標マッピング": [
        { "指標": "実数",                "列オフセット": 0 },
        { "指標": "対前年同月-増減",      "列オフセット": 1 },
        { "指標": "対前年同月-増減率(%)", "列オフセット": 2 }
      ]
    }
    $$) AS "設定JSON",
    TRUE AS "有効フラグ"
) s
ON t."テンプレID" = s."テンプレID"
WHEN MATCHED THEN UPDATE SET
  t."テンプレ名" = s."テンプレ名",
  t."適用条件_ファイル名正規表現" = s."適用条件_ファイル名正規表現",
  t."命名規則_説明" = s."命名規則_説明",
  t."命名規則_正規表現" = s."命名規則_正規表現",
  t."実行SP名" = s."実行SP名",
  t."設定JSON" = s."設定JSON",
  t."有効フラグ" = s."有効フラグ",
  t."更新日時" = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT (
  "テンプレID",
  "テンプレ名",
  "適用条件_ファイル名正規表現",
  "命名規則_説明",
  "命名規則_正規表現",
  "実行SP名",
  "設定JSON",
  "有効フラグ"
) VALUES (
  s."テンプレID",
  s."テンプレ名",
  s."適用条件_ファイル名正規表現",
  s."命名規則_説明",
  s."命名規則_正規表現",
  s."実行SP名",
  s."設定JSON",
  s."有効フラグ"
);

すべて実行完了後、プロシージャの作成と取込テンプレが登録されていることを確認してください。

4. 手動実行確認

ここまでの作業で「環境」と「処理部分」は作成できているので、実際に手動で動くかどうかを確認しておきます。

まずは、「取込ステージ」にファイルが配置されているかを下記コードで確認します。

USE ROLE "TEST_ROLE";
USE WAREHOUSE SAMPLE_WH;
USE DATABASE "SAMPLE_DB";
USE SCHEMA "SAMPLE_SCHEMA";

LIST '@"取込ステージ"';

※上記のLISTコマンドで取得した結果を次のSQL実行（RESULT_SCAN）でそのまま活用するので、必ず実行してください。

実行結果にステージへアップロードしたファイルが出力されていればOKです。

次に、LISTコマンドで出力されたステージ上のファイルを「原本ファイル管理」に登録するため、下記コードを実行します。

MERGE INTO "原本ファイル管理" t
USING (
  SELECT
    CONCAT('@', "name") AS "ステージパス",
    REGEXP_SUBSTR("name", '[^/]+$') AS "ファイル名",
    LOWER(REGEXP_SUBSTR("name", '[^.]+$')) AS "拡張子"
  FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()))
) s
ON t."ステージパス" = s."ステージパス"
WHEN NOT MATCHED THEN
  INSERT ("ステージパス","ファイル名","拡張子","ステータス")
  VALUES (s."ステージパス", s."ファイル名", s."拡張子", '未処理');

実行結果は以下となります。

次に、「原本管理ファイル」の中身を見て、先ほど登録した「未処理」の原本ファイルを確認し、次のSQL実行で使用する 原本ファイルID を控えておくために下記コードを実行します。

SELECT *
FROM "原本ファイル管理"
ORDER BY "登録日時" DESC;

実行結果を確認し、原本ファイルID（今回は「1」）を控えます。

あとは、原本ファイルIDを指定する形でプロシージャを手動実行して、構造化処理を試してみます。

下記コードを実行します。
※()内に控えておいた原本ファイルIDを記載します。

CALL "SP_労働力調査_月次積上_取込"(1);

実行結果を確認し、結果が「成功」となっていることを確認します。

実際に、出力テーブルである「TBL_労働力調査_月次積上」を確認するために下記コードを実行します。

SELECT * FROM SAMPLE_DB.SAMPLE_SCHEMA."TBL_労働力調査_月次積上";

実行結果を確認すると、複雑だったデータが縦持ちの使いやすい形で構造化されたデータが入っていることを確認できました。

提供ビューである「VIE_労働力調査_月次積上」もSELECTして確認しましたが、おそらく問題なく使えそうです。

ここまでの検証で、手動で非構造データを構造データとして取り込むところまでは確認ができましたので、いよいよこれらの操作（手動で実行確認したような内容）をGUIでできるようにするための作業に入っていきます。

5. アプリケーション作成

実際にデータ管理担当者がGUIベースで構造化処理を行えるように、アプリケーションを作成していきます。

ここで活躍するのがStreamlit in Snowflakeになります。

Streamlit in Snowflakeを簡単に説明すると、

StreamlitはオープンソースのPythonライブラリで、カスタムウェブアプリを簡単に作成できる
そのStreamlitをSnowflake上で安全の構築・展開・共有することができるようにしたものがStreamlit in Snowflake

といった感じです。

アプリケーション作成を進めるために、まずは自身のロールを検証用に作成した「TEST_ROLE」に切り替えます。

この状態で、「プロジェクト」から「Streamlit」を押下します。

右上の「＋Streamlitアプリ」を押下します。

任意のアプリタイトルを決め、アプリの場所を今回作成した「SAMPLE_DB.SAMPLE_SCHEMA」に設定し、Python環境は「ウェアハウスで実行」、アプリウェアハウスは「SAMPLE_WH」にして「作成」を押下します。

すると、アプリが作成され、最初からサンプルとして入っているアプリが表示されます。

左側に2つのファイルが表示されていますが、今回は「streamlit_app.py」の中身をいじくっていきたいと思います。

それぞれの役割については、以下となります。

~.yml：アプリ実行環境の定義ファイル
~.py：アプリ本体のコード

元々入っていたソースコードを削除し、新たに作成したPythonベースのソースコードを貼り付けて「実行」を押下します。

実行完了後、今回作成したいアプリが無事に表示されました。

6. 触ってみる

最後に、作成した「非構造データ取込アプリ」を実際に触っていきます。

予め、「取込ステージ」に複数のファイルをアップロードしておきます。

アプリに戻り、まずは「Step1 事前確認」を行っていきます。

投入したファイルがルールに則っているかのチェックと、取込テンプレの選択を行い、次のステップに進みます。

2つのチェックを入れると、次のステップに進むことができます。

「Step2 ファイル登録」では、ステージにあるファイルを選択し、原本登録していきます。

「取込ステージ」にある最新のファイル情報を取得するために、「ステージ一覧を更新」ボタンを押下し、どんなファイルがあるかを確認します。

ボタン押下後、「ステージ上にどのくらいファイルがあるか」「それぞれのファイルの登録状況」「ファイルの命名規則判定」を簡単に見ることができます。

また、未登録ファイルと登録済みファイルをタブで切り替えながら詳細な中身を確認することもできます。

登録済みファイルを見ると、先ほど手動で実行確認するために原本ファイル管理に登録していた2025年1月分のファイルがありました。

今回は、新しく投入したファイルを対象にデータ取込を行うので、アプリ下部にある未処理のファイルを選択できるところから、任意のファイルを選択して登録していきます。

2025年2月分を指定し、「台帳に登録する（1件）」を押下し登録します。

※アプリ自体は複数選択できる仕様にしたので、一括でステージ上の複数ファイルを原本登録することもできました。

次は、いよいよ「Step3 取込実行」をやっていきます。

先ほど登録した2025年2月分のファイルが「未処理のファイル」として一覧に表示されています。

アプリ右側の「実行対象ファイル」を選択することができるので、当該ファイルを指定し、「取込を実行する」ボタンを押下しすると、取込処理が走ります。

10秒ほど経過した後に、実行結果が表示されました。

無事に成功したようです。

※原本登録と同様に、取込処理についても複数の未処理ファイルを一度に選択して実行することができるようにしたのですが、これもうまく動作しました。

これだけではまだ不安が残るので、最後に「Step4 結果確認」に進み、取込処理後のデータをプレビューにて確認していきます。

「出力プレビュー」にて、先ほど実行した2025年2月分の取込結果を見ると、レコードとしては476件なので問題なさそうです。

また、右側の出力データサンプルもざっと確認し、問題なくデータが入っていそうでした。

そして、ここまで一切触れていなかった「取込実行ログ」と「取込エラーログ」についても見ていきます。

「取込実行ログ」を確認すると、これまでに試行した2回分のログが溜まっていました。

「取込エラーログ」はない状態です。

試しに、今回のフォーマットとは異なるExcelファイルを命名規則だけ守った状態（2025年5月分として）でステージにアップロードし、取込処理を試してみました。

「Step1 事前確認」を早々に済ませ、「Step2 ファイル登録」にて2025年5月分のファイルを選択し、登録します。

次に、当該ファイルを指定した形で取込処理を実行したところ・・・

きちんと「失敗」の結果が出てくれました。
※以下イメージでは取込処理前後で原本ファイルIDが異なっておりますが、裏側でいろいろ修正をしただけになりますので、結果自体に影響はありません。

また、「Step4 結果確認」に移り、「取込エラーログ」にも記録されていたので、ここで取込に失敗したファイルを確認し、元データ確認などの対象ファイルを追うこともできました。

これまでの一連の流れ（Step1～Step4）について、手動でステージにファイルをアップロードした状態から始めると、1ファイルの登録には体感1分もかけずにボタンをポチポチするだけで実行することができるようになりました。

さいごに

Snowpark × Streamlit in Snowflakeを活用し、GUI操作で非構造データのデータ取込（構造化）ができるようなアプリケーション/仕組みの作成を実施しました。

検証してみた一番の感想としては、

決まった様式（データフォーマット）であれば、かなり便利に構造化できるのでは？

と感じました。

ただ、逆に「様式が可変で、毎回中身を見て判断しなければならないもの」については、今回のような仕組みよりも、Snowflakeが公式に提供している機能（Cortex AI Documents）や、AI読み取り機能に強みのあるETL製品等が活躍しそうな気がします。

Cortex AI 関数:ドキュメント | Snowflake Documentation

また、この仕組みを発展させて、

ステージへのファイルアップロードを自動化
処理自体をスケジューリング実行できるようにしておいて、データ管理担当者は実行結果だけを確認

のようなこともできれば、さらに活用しやすい仕組みになりそうです。

取組背景でも記載した通り、「扱いづらいけれども、是非ともデータ化して活用したい」と思えるようなデータは、実はたくさんあると思っています。

そんな眠れるお宝（データ）を拾い上げるために、非常に有意義な検証ができました。

本記事を通して、上記のようなお悩みを持たれていた方々の作業負担が少しでも軽減されれば幸いです。

実際に試してみた感想やご質問などがあれば、是非ともコメントお待ちしております。