ユーザーがファイルをSnowflakeにアップロードする方法：part1.5 S3＋Snowpipeにバリデーションを加えよう編
SQLをふだん書かないユーザーが、手元のファイルをSnowflakeのテーブルに取り込む方法を、シリーズで一つずつ試していきます。
part1.5としては、前回構築した、ユーザーにcsvをS3へ置いてもらい、それをSnowpipeで取り込むという方法を改良し、S3に置いたファイル自体のバリデーションを仕込もうと思います。
記事3行概要
- Snowpipeの手前にLambdaを1つ置くと、壊れたCSVをテーブルに入れる前に弾いて、理由をメールでユーザーに差し戻せるよ！
- 検証OKのファイルは、名前を変えてからSnowpipeに渡さないといけないよ！Snowpipeは同じ名前のファイルを、中身が変わっていても再ロードしないよ！
- 名前に混ぜるのはタイムスタンプではなく元ファイルのeTagだよ！同じファイルでLambdaが2回動いても二重ロードにならないよ！
前回のおさらいと、今回足すもの
前回作ったのは、S3にCSVを置くとSnowflakeのテーブルに入る、という一方通行の経路でした。
ユーザーがs3://<my_bucket>/load/files/にCSVを置くと、S3のイベント通知がSQS経由でSnowpipeに届き、パイプ（mypipe）がmytableへロードします。
この構成では、置かれたファイルの中身を誰も見ていません。
ヘッダの列名がずれたCSV、列数が足りない行が混ざったCSV、Excelの既定設定で保存してShift_JISになったCSVでも、そのままロードが走ります。
Snowpipeがまったくの無防備というわけではありません。
COPY INTOのON_ERRORは、バルクロードではABORT_STATEMENTが既定値ですが、SnowpipeではSKIP_FILEが既定値です。
エラー行が見つかったファイルは、ファイルごとロードされずに飛ばされます。
問題は、飛ばしたことがユーザーに伝わらない点です。
ユーザーはS3に置いた時点で作業を終えたと思っていて、テーブルに入っていないことに気づくのは、誰かがデータを使おうとしたときです。
しかも気づいたのがエンジニア側だと、原因のファイルを特定してユーザーに連絡する、という手戻りが発生します。
そこで今回は、ロードする前にファイルの形式を検証する門番を1つ足して、結果をメールで返すようにします。
検証を通ったファイルはこれまで通りSnowpipeに流し、通らなかったファイルは隔離して、理由を付けてユーザーに差し戻します。
今回の全体像
S3のプレフィックスを3つに分ける
門番を置くために、ユーザーが置く場所とSnowpipeが読む場所を分けます。
| プレフィックス | 書く人 | 読む人 | 役割 |
|---|---|---|---|
incoming/ |
ユーザー | Lambda | 受付。ユーザーが触るのはここだけ |
load/files/ |
Lambda | Snowpipe | 検証済み。part1のステージとパイプがそのまま読む |
rejected/ |
Lambda | ユーザー | 差し戻し。理由を書いたテキストを同梱する |
ユーザーの投入先がload/files/からincoming/に変わるだけで、Snowflake側のステージ、パイプ、ストレージ統合、IAMロールはpart1のまま使えます。
この3つを、範囲が重ならない兄弟プレフィックスにしておく理由が2つあります。
1つは、S3が同一バケット内で範囲の重なるイベント通知を作れないためです。
part1ではプレフィックスを指定せず、バケット全体をSnowpipe向け通知の対象にしていました。
このままLambda向けの通知を足そうとすると、範囲が重なって作成できません。
もう1つは、LambdaがS3へ書き戻すためです。
トリガの範囲と書き込み先が重なると、Lambdaが自分の出力で自分を起動します。
トリガはincoming/だけに付け、書き込み先は必ず別のプレフィックスにします。
何をどこで検証するか
検証項目は、ファイルだけを見て判定できるものと、ほかのデータと突き合わせないと判定できないものに分かれます。
今回Lambdaに持たせるのは前者だけです。
| レイヤ | 見るもの | 置く場所 |
|---|---|---|
| L0 受付 | 拡張子が.csv／0バイトでない／サイズ上限内 |
Lambda |
| L1 文字コード | UTF-8でデコードできる（BOMは除去する） | Lambda |
| L2 構造 | ヘッダが完全一致／全行の列数が一致／データ行が1行以上／必須列が空でない | Lambda |
| L3 型と業務ルール | 日付や数値としてパースできる／コード値がマスタに存在する | Snowflake側（今回は扱わない） |
L3をLambdaへ寄せると、マスタのコピーをLambdaに持たせることになります。
マスタが更新されたときに検証ロジックとテーブルの二重管理になるので、L3はSnowflake側の仕事として切り離しました。
Snowflake側でどこまで見られるかは、それ自体で1本になる分量なので次回以降に回します。
「必須列が空でない」をL2に置いているのは、空かどうかがファイル1つだけで判定できるためです。
値の妥当性を見る点ではL3に近いのですが、コード値の検証と違ってマスタが要りません。
ファイル単体で判定できるものはLambdaに寄せる、という切り分けに従ってこちらに入れています。
文字コードの扱いは、先に方針を決めておきます。
- BOM付きUTF-8（Excelでファイル形式に「CSV UTF-8」を選んだ場合）は、BOMを除去して通す
- Shift_JISやCP932（Excelで「CSV」を選んだ場合）は、通さずに差し戻す
Shift_JISを自動でUTF-8へ変換する作りにもできますが、今回は選びませんでした。
変換に失敗した文字は多くの場合そのままテーブルに入り、後から見つけるのが難しくなります。
「UTF-8で保存し直してください」と返して、ファイルを直してもらうほうが事故が小さく済みます。
手順の一覧
実際に手を動かすのはStep 0からStep 5までで、すべてAWS側です。
- Step 0. part1のイベント通知に
load/files/のプレフィックスを付ける（AWS） - Step 1.
incoming/とrejected/を用意する（AWS） - Step 2. 通知用のSNSトピックを作り、メールを購読する（AWS）
- Step 3. Lambdaの実行ロールを作る（AWS）
- Step 4. Lambda関数を作る（AWS）
- Step 5.
incoming/にイベント通知を設定してLambdaを起動する（AWS） - Step 6. パイプにロードエラーの通知先を付ける（Snowflake、今回はスキップ）
part1がSnowflakeとAWSを何度も往復したのに対して、今回はAWS側だけで完結します。
Step 6はSnowflake側の保険で、今回は実施していません。
手順だけ最後に置いておきます。
検証を通ったファイルは、名前を変えてから渡す
手順に入る前に、Step 4の作りを決めている制約を1つ確認します。
これを踏まえないと、差し戻しの導線がつながったように見えて実際には壊れます。
Snowpipeは、パイプごとのロード履歴メタデータにファイルのパスと名前を記録し、同じファイルを二重にロードしないようにしています。
公式ドキュメントの記述は次の通りです。
This metadata stores the path (i.e. prefix) and name of each loaded file, and prevents loading files with the same name even if they were later modified (i.e. have a different eTag).
判定に使われるのはパスと名前だけで、中身が変わっていても、同じ名前なら再ロードされません。
part1でも触れた挙動ですが、差し戻しを作ると影響が変わります。
具体的には、次の順番で起きます。
- ユーザーが
sales.csvを置く。形式は正しいのでLambdaは通し、Snowpipeがロードする - あとで値の誤りに気づき、ユーザーが直して同じ名前で置き直す
- Lambdaは形式を見ているだけなので、当然これも通す
- Snowpipeは「その名前は済み」と判断してロードしない
- ユーザーには受付OKのメールだけが届き、直したはずのデータはテーブルに入らない
修正版が黙って捨てられるうえ、通知は成功と伝えてきます。
差し戻しの仕組みを入れる以上、直したファイルが確実に入ることまで面倒を見る必要があるので、load/files/へコピーするときに名前を変えます。
タイムスタンプではなくeTagを混ぜる
名前を一意にするだけなら、コピー時刻を付けるのが手軽です。
ただ、この用途では合いません。
S3のイベント通知は少なくとも1回の配信で、同じファイルについてLambdaが2回起動することがあります。
処理時刻から名前を作ると、1回目と2回目で違う名前になり、同じ中身が2つのファイルとしてload/files/に並びます。
Snowpipeから見れば別名なので、どちらもロードされて重複します。
そこで、元オブジェクトのeTagを名前に混ぜます。
load/files/<元のファイル名>__<eTagの先頭8桁>.csv
eTagはオブジェクトごとに決まる値なので、次の2つが同時に成り立ちます。
- 同じオブジェクトでLambdaが再実行された場合：コピー先の名前も同じになる。Snowpipeの重複防止がそのまま二重ロードを止める
- ユーザーが中身を直して置き直した場合：eTagが変わるので別名になる。Snowpipeから見て未処理のファイルなのでロードされる
Snowpipeの重複防止は、名前を変えて避ける対象として扱われがちですが、名前の付け方を揃えれば冪等性の担保として使えます。
再送に備えた重複排除の仕組みを自前で持たなくて済むので、今回はこちらに寄せました。
手順
Step 0. part1のイベント通知をload/files/に絞る（AWS）
S3コンソールで対象バケット → プロパティ → イベント通知と進み、part1で作ったSQS向けの通知を編集します。
プレフィックスにload/files/を入れて保存します。
part1のときはプレフィックスが空で、バケット全体が対象でした。
この状態のままStep 5でincoming/向けの通知を作ろうとすると、範囲が重なって作成できません。
なお、incoming/を別のバケットに分ける手もあります。
同一バケットではないので範囲は重なりませんし、ユーザーに見せるバケットと内部用のバケットを分けられます。
今回はpart1の構成からの差分を小さくしたかったので、1つのバケットをプレフィックスで住み分ける形にしました。
Step 1. incoming/とrejected/を用意する（AWS）
part1のバケットに、受付用と隔離用のフォルダを作ります。
s3://<my_bucket>/incoming/s3://<my_bucket>/rejected/
rejected/は放置すると溜まり続けるので、ライフサイクルルールで一定期間後に削除するようにしておきます。
差し戻したファイルは、ユーザーが直して置き直せばそれ以上使いません。
Step 2. 通知用のSNSトピックを作り、メールを購読する（AWS）
検証結果を送るためのSNSトピックを作り、通知を受け取るメールアドレスをサブスクリプションに登録します。
- Amazon SNS → トピック → 「トピックの作成」（タイプはスタンダード）
- 作成したトピック → 「サブスクリプションの作成」→ プロトコルにEメール、エンドポイントに宛先アドレス
- 宛先に確認メールが届くので、リンクを開いて購読を確定する
購読が確定するまで通知は届きません。
ユーザーごとにアドレスを登録するなら、この確認作業はユーザー自身にやってもらう必要があります。
もう1つ、SNSのPublishはSubjectをASCIIのみと定めています。
日本語の件名は渡せないので、件名は英数字にして、日本語はすべて本文に入れます。
この2点があるので、業務ユーザーに配る通知としては、SNSは仮組み向きです。
件名も本文も自由に作りたい、差出人を自社のドメインにしたい、といった要求が出たらSESに移すことになります。
今回は、Slackでメールを受け取れるようにしています。
チャンネル設定からそのチャンネルのアドレスを取得できるので、そこへ送る形にしています。
(筆者はテストしている途中で、メールのConfirmをしていなかったことに気づきましたよ！ちゃんとConfirmしましょうね！)
Step 3. Lambdaの実行ロールを作る（AWS）
Lambdaがファイルを読み書きして通知を送れるよう、実行ロールを作ります。
part1でSnowflakeに引き受けさせたロールとは別物です。
あちらはSnowflakeがS3を読むために使うロール、こちらはLambda自身が使うロールです。
必要な権限は4つです。
-
incoming/からの読み取りと削除 -
load/files/とrejected/への書き込み - 通知用SNSトピックへの
Publish - CloudWatch Logsへの書き込み（
AWSLambdaBasicExecutionRoleで付く）
実行ロールに付けるIAMポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:DeleteObject"],
"Resource": "arn:aws:s3:::<my_bucket>/incoming/*"
},
{
"Effect": "Allow",
"Action": ["s3:PutObject"],
"Resource": [
"arn:aws:s3:::<my_bucket>/load/files/*",
"arn:aws:s3:::<my_bucket>/rejected/*"
]
},
{
"Effect": "Allow",
"Action": ["sns:Publish"],
"Resource": "<通知用SNSトピックのARN>"
}
]
}
load/files/とrejected/へのs3:PutObjectは、LambdaがCopyObjectでファイルを移すために要ります。
CopyObjectはコピー元のs3:GetObjectとコピー先のs3:PutObjectの両方を要求しますが、どちらが欠けてもエラーはCopyObjectのAccessDeniedとして出ます。
足りない側は、メッセージ中のアクション名とリソースのARNで見分けます。
incoming/にs3:DeleteObjectを付けているのは、検証が済んだファイルを受付から消すためです。
消さずに残すと、次にユーザーが同じ名前で置いたときに、どちらが処理済みなのか判別できなくなります。
(筆者はResourceがarn:aws:s3:::arn:aws:s3:::<my_bucket>/load/files/*という二重の形になっていることに気づかず、だいぶ時間を取られました。)
Step 4. Lambda関数を作る（AWS）
incoming/に置かれたCSVを検証する関数を作ります。
ランタイムはPython、環境変数NOTIFY_TOPIC_ARNにStep 2のトピックARNを入れる前提です。
実行ロールは、関数の作成画面で「既存のロールを使用する」を選び、Step 3で作ったロールを指定します。
既定は「基本的な Lambda アクセス権限で新しいロールを作成」で、こちらを選ぶとCloudWatch Logsにしか書けないロールが付くので、最初のS3読み取りでAccessDeniedになります。
コードは、関数を作ったときに生成されるlambda_function.pyへそのまま貼り付けます。
エントリポイントの関数名をlambda_handlerにしてあるので、ランタイム設定のハンドラは既定のlambda_function.lambda_handlerのままで動きます。
この設定値は<ファイル名>.<関数名>という形で、どちらか一方でも実物とずれていると、実行時にRuntime.HandlerNotFoundで落ちます。
(権限的にdenyされた場合には、Lambdaの実行ロールが想定したものになっているかを確認しましょう。)
環境変数は、設定 → 環境変数 → 編集から追加できます。
Lambdaのコードで検証するのは、part1のダミーデータに合わせたL0からL2までです。
ヘッダは「名称,アップロード方法」の2列を想定し、どちらも空を許さない必須列として扱います。
最初に見つかった問題で打ち切るので、1つのファイルに問題が複数あるときは、直して置き直すたびに次の問題が返ります。
(コードはClaudeくんに作らせました。)
Lambdaコードと説明(by Claude)
import csv
import io
import os
import urllib.parse
import boto3
s3 = boto3.client("s3")
sns = boto3.client("sns")
EXPECTED_HEADER = ["名称", "アップロード方法"] # part1のダミーデータに合わせる
REQUIRED_COLUMNS = ["名称", "アップロード方法"] # 空を許さない列。今回は2列とも必須
INCOMING_PREFIX = "incoming/" # 受付
LOAD_PREFIX = "load/files/" # 検証OK。part1のSnowpipeが読む
REJECT_PREFIX = "rejected/" # 検証NG。隔離先
MAX_BYTES = 10 * 1024 * 1024 # L0のサイズ上限。超えたら中身を読まずに差し戻す
TOPIC_ARN = os.environ["NOTIFY_TOPIC_ARN"]
class Rejected(Exception):
"""差し戻す理由を持たせた例外"""
def validate(body: bytes) -> None:
"""L1とL2の検証。問題があればRejectedを投げる"""
try:
text = body.decode("utf-8-sig") # BOM付きUTF-8はBOMを落として読む
except UnicodeDecodeError:
raise Rejected(
"UTF-8で読めませんでした。"
"Excelで保存する場合は、ファイル形式に「CSV UTF-8」を選んでください。"
)
rows = list(csv.reader(io.StringIO(text)))
if not rows:
raise Rejected("中身が空です。")
if rows[0] != EXPECTED_HEADER:
raise Rejected(
f"1行目のヘッダが想定と違います。想定={EXPECTED_HEADER} 実際={rows[0]}"
)
if len(rows) == 1:
raise Rejected("ヘッダ行しかなく、データ行がありません。")
for lineno, row in enumerate(rows[1:], start=2):
if len(row) != len(EXPECTED_HEADER):
raise Rejected(
f"{lineno}行目の列数が{len(row)}です。{len(EXPECTED_HEADER)}列にしてください。"
)
for col, value in zip(EXPECTED_HEADER, row):
if col in REQUIRED_COLUMNS and not value.strip(): # 空白だけのセルも空とみなす
raise Rejected(f"{lineno}行目の「{col}」が空です。")
def promote(bucket: str, key: str, filename: str, etag: str) -> str:
"""検証OK。eTagを混ぜた名前でSnowpipeが読むフォルダへ置く"""
stem, _, ext = filename.rpartition(".")
dest = f"{LOAD_PREFIX}{stem}__{etag[:8]}.{ext}"
s3.copy_object(Bucket=bucket, Key=dest, CopySource={"Bucket": bucket, "Key": key})
return dest
def quarantine(bucket: str, key: str, filename: str, reason: str) -> str:
"""検証NG。隔離して、理由を同じ場所にテキストで残す"""
dest = f"{REJECT_PREFIX}{filename}"
s3.copy_object(Bucket=bucket, Key=dest, CopySource={"Bucket": bucket, "Key": key})
s3.put_object(Bucket=bucket, Key=f"{dest}.error.txt", Body=reason.encode("utf-8"))
return dest
def lambda_handler(event, context):
rec = event["Records"][0]
bucket = rec["s3"]["bucket"]["name"]
key = urllib.parse.unquote_plus(rec["s3"]["object"]["key"])
filename = key.rsplit("/", 1)[-1]
size = rec["s3"]["object"]["size"]
etag = rec["s3"]["object"]["eTag"].strip('"')
copied = False # incoming/の外へ退避できたか
try:
# L0: 中身を読む前に判断できるもの
if not filename.lower().endswith(".csv"):
raise Rejected("拡張子が.csvではありません。")
if size == 0:
raise Rejected("0バイトのファイルです。")
if size > MAX_BYTES:
raise Rejected(
f"サイズが上限の{MAX_BYTES}バイトを超えています。ファイルを分割してください。"
)
try:
body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
except s3.exceptions.NoSuchKey:
return {"skipped": filename} # イベントの再配信。1回目の実行で処理済み
validate(body) # L1とL2
dest = promote(bucket, key, filename, etag)
copied = True
subject = "[csv-validator] accepted"
message = (
f"ファイル: {filename}\n"
f"検証結果: OK\n"
f"取り込み待ち: s3://{bucket}/{dest}\n\n"
"数分後にSnowflakeのテーブルへ反映されます。"
)
except Rejected as e:
dest = quarantine(bucket, key, filename, str(e))
copied = True
subject = "[csv-validator] rejected"
message = (
f"ファイル: {filename}\n"
f"検証結果: NG\n"
f"理由: {e}\n\n"
f"直したファイルを s3://{bucket}/{INCOMING_PREFIX} に置き直してください。\n"
f"元のファイルは s3://{bucket}/{dest} に退避しています。"
)
except Exception as e:
# 検証NG以外の失敗でも、退避と通知はする
subject = "[csv-validator] error"
message = f"ファイル: {filename}\n検証処理が異常終了しました。\n{e!r}"
try:
dest = quarantine(bucket, key, filename, f"検証処理が異常終了しました: {e!r}")
copied = True
message += f"\n退避先: s3://{bucket}/{dest}"
except Exception as qe:
# 隔離が失敗しても通知は出す。ファイルはincoming/に残す
message += f"\n退避にも失敗しました: {qe!r}"
print(f"{subject}\n{message}") # どの経路を通ったかをCloudWatch Logsに残す
sns.publish(TopicArn=TOPIC_ARN, Subject=subject, Message=message)
if copied:
s3.delete_object(Bucket=bucket, Key=key) # 退避できたときだけ受付から消す
return {"file": filename, "subject": subject, "copied": copied}
素直に書くと落ちる箇所を4つ手当てしてあるので、それぞれ補足します。
1つ目は、get_objectでNoSuchKeyを捕まえて何もせず返している箇所です。
イベントの再配信で2回目が動いたとき、1回目が削除まで終えているとファイルはもう無いので、これは異常ではありません。
ここを一般の例外として扱うと、正常に処理できたファイルについて検証処理エラーのメールが飛びます。
2つ目は、Rejectedとは別にExceptionも受けてquarantineへ流している箇所です。
Lambdaがそのまま落ちるとファイルはincoming/に残り、通知も飛びません。
ユーザーは置いた時点で作業を終えているので、この状態は誰にも気づかれずに止まります。
理由が分からない場合でも、隔離して通知するところまでは通しておきます。
このquarantineをさらにtryで包んでいるのは、隔離そのものが失敗しうるためです。
たとえば実行ロールにs3:GetObjectが付いていないと、最初のget_objectがAccessDeniedで失敗し、続くquarantineのcopy_objectもコピー元の読み取りで同じ理由で失敗します。
包まずに書くと、Lambdaが返すのは後から起きたCopyObjectの失敗だけになり、元の原因が隠れます。
一方、差し戻し経路のquarantineは包んでいません。
ここが落ちた場合はLambdaごと落ちるので、ファイルはincoming/に残り、実行イベントはデッドレターキューに残ります。
3つ目は、退避できたときだけdelete_objectを呼ぶようにしている箇所です。
incoming/にあるのはユーザーが置いた1部だけなので、どこにもコピーできていない状態でここを消すと、そのファイルが失われます。
受付に残しておけば、原因を直したあとに置き直しを頼まずに済みます。
4つ目は、MAX_BYTESでサイズを見てからget_objectしている箇所です。
このコードは中身を一度メモリに読むので、大きいファイルではLambdaのメモリと実行時間の上限に当たります。
上限を超えるファイルを扱う必要が出たら、行ごとのストリーミング処理に書き換えることになります。
通知の直前にprintを1行入れて、どの経路を通ったかをCloudWatch Logsに残しています。
この関数はOKでもNGでも同じように正常終了し、incoming/のファイルを消すので、ログがないと結果を外から見分けられません。
メールが届かないときに、検証で弾かれたのか、通知の側で止まっているのかを、ログだけで切り分けられるようにしておきます。
Step 5. incoming/にイベント通知を設定してLambdaを起動する（AWS）
incoming/にファイルが置かれたらLambdaが動くよう、イベント通知を設定します。
- 対象バケットのプロパティ → イベント通知 → 「イベント通知を作成」
-
プレフィックス：
incoming/ -
イベントタイプ：
s3:ObjectCreated:* - 送信先：Lambda関数（Step 4で作った関数）
Step 0でSnowpipe向けの通知をload/files/に絞ってあるので、この通知とは範囲が重なりません。
Step 6. パイプにロードエラーの通知先を付ける（Snowflake、今回はスキップ）
ここまでで、形式が壊れたファイルはテーブルに入りません。
一方で、L3にあたる型の不一致は素通りするので、ロード時にSnowpipe側でエラーになります。
既定のSKIP_FILEで飛ばされるだけだと気づけないため、本来ならパイプにエラー通知の送信先を付けておきたいところです。
今回は実施していないので、やり方だけ残しておきます。
パイプのERROR_INTEGRATIONに通知統合を指定すると、ロードのエラーをSNSなどのメッセージングサービスへ送れます。
CREATE OR REPLACE PIPE snowpipe_db.public.mypipe
AUTO_INGEST = TRUE
ERROR_INTEGRATION = <通知統合名> -- 通知統合はあらかじめ作っておく
AS
COPY INTO snowpipe_db.public.mytable
FROM @snowpipe_db.public.mystage;
既存のパイプへ後から付けるならALTER PIPEもありますが、構文と制限はドキュメントを確認してください。
通知統合の作成にはSnowflakeがSNSへPublishするためのIAM設定が別途必要で、Step 2で作ったユーザー向けのトピックとは別に用意することになります。
ここは今回試せていないので、公式ドキュメントへのリンクだけ置いておきます。
動作確認
正常系と異常系をincoming/に置いて、想定通りに振り分けられるかを見ます。
| ケース | 置くファイル | 期待する結果 |
|---|---|---|
| 正常系 | ヘッダと列数が正しいUTF-8のCSV |
load/files/へ昇格し、mytableに入る。受付OKのメールが届く |
| 異常系1 | ヘッダの列名を崩したCSV |
rejected/へ隔離。差し戻しメールが届き、テーブルは変化しない |
| 異常系2 | Shift_JISで保存したCSV | 同上。理由が文字コードになる |
| 異常系3 | 必須列が空のセルを含むCSV | 同上。理由に行番号と列名が入る |
| 再投入1 | 異常系1を直して同じ名前で置き直す |
load/files/へ昇格してロードされる |
| 再投入2 | 正常系のファイルの値を1つ書き換えて、同じ名前で置き直す | 別名で昇格し、書き換えた行がテーブルに入る |
正常系
| 名称 | アップロード方法 |
|---|---|
| apple | s3_normal |
| blueberry | s3_normal |
| cherry | s3_normal |
| date | s3_normal |
異常系1
| 名称 | アップロード方法_X |
|---|---|
| apple | 名称ずれ |
| blueberry | 名称ずれ |
| cherry | 名称ずれ |
| date | 名称ずれ |
異常系2(Shift_JIS保存)
| 名称 | アップロード方法 |
|---|---|
| apple | s3 |
| blueberry | s3 |
| cherry | s3 |
| date | s3 |
異常系3(必須列が空)
| 名称 | アップロード方法 |
|---|---|
| apple | s3 |
| blueberry | |
| cherry | |
| date |
再投入1(ヘッダ名修正)
| 名称 | アップロード方法 |
|---|---|
| apple | 名称ずれ |
| blueberry | 名称ずれ |
| cherry | 名称ずれ |
| date | 名称ずれ |
再投入2
| 名称 | アップロード方法 |
|---|---|
| apple | s3_retry |
| blueberry | s3_retry |
| cherry | s3_retry |
| date | s3_retry |
前半の4ケースは振り分けの確認で、最後の2つは差し戻しの導線が一周するかの確認です。
このうち、名前にeTagを混ぜた設計が効いているかを判定できるのは再投入2だけです。
再投入1で置き直すファイルは、異常系1で隔離されていてload/files/に届いていません。
Snowpipeはそのパスと名前を知らないので、名前をそのままコピーする作りでもロードされます。
これに対して再投入2で置き直すファイルは、正常系で一度ロードを終えています。
名前を変えずにコピーする作りだと、Snowpipeが同名と判断してこのケースだけが失敗します。
動作確認の結果
正常系の時刻は次の通りです（CloudWatchのUTC表記はJSTに直しています）。
- Lambda起動：2026-08-02 21:05:33 JST
- S3のファイル置き換え：2026-08-02 21:05:34 JST
異常系1エラーメッセージ
異常系2エラーメッセージ
異常系3エラーメッセージ
再投入1成功メッセージ
再投入2成功メッセージ
投入後S3
投入後Snowflake(アップロード方法カラムのみ抽出)
以上から、意図した通りに動いたことを確認できました。
(一部nullが入っていますが、見なかったことにしてください。(一回ミスりました。))
今回のコストについて
part1の構成に対して増えるのは、次の3つです。
- Lambdaの実行（ファイル1つあたり1回、実行時間は数百ミリ秒程度の想定）
- S3のリクエスト（
GetObject、CopyObject、PutObject、DeleteObject） - SNSのメール送信（ファイル1つあたり1通）
part1ではSnowpipeが0.000000000クレジット、S3が1.95e-12ドルという、桁が見えない結果でした。
Lambdaの料金は以下のサイト記載の通りです。
今回、一回の実行結果としては、512ミリ秒、128MBメモリでしたので、1ミリ秒あたり0.0000000021ドルより、0.0000010752ドルとなりました。
これとは別にリクエスト料金が100万件あたり0.20ドル、つまり1回あたり0.0000002ドルかかります。
よって、今回もほぼ0と言っていいでしょう。
(ただし、Lambdaがエラーで終わると、非同期呼び出しのリトライでファイル1つにつき最大3回実行されます。失敗が続くとその分だけ実行時間が積み上がるので、注意してください。)
本手法のユーザー目線でのメリデメ
part1と同じ観点で並べます。
| 対象 | メリット | デメリット |
|---|---|---|
| ユーザー | ミスの理由がメールで返るので、自分で直して出し直せる。エンジニアへの問い合わせが要らない | S3を触る必要は変わらない。SNSの購読確認という初回作業が増える |
| エンジニア | 壊れたファイルがテーブルに入らない。原因ファイルの特定と連絡から解放される | Lambda、SNS、IAMロールが増える。ヘッダや列の定義が変わるたびに検証ロジックの保守が要る |
part1の表と見比べると、埋まったのはユーザー側のデメリットのうち「上げたファイルがミスっていた場合の対応がユーザー側で完結しない」という点だけです。
「csvフォーマットを調整してS3にアクセスしないといけない」というハードルの高さは、今回の変更では下がっていません。
結論
Snowpipeの手前にLambdaを1つ挟むと、形式が壊れたファイルをテーブルに入れる前に弾いて、理由を付けてユーザーに差し戻せます。
part1で作ったストレージ統合、ステージ、パイプはそのまま使えて、Snowflake側の作り直しは要りませんでした。
作業の山場はLambdaのコードではなく、検証を通ったファイルの名前の付け方でした。
Snowpipeはパスと名前で重複を判定するので、そのままコピーすると、ユーザーが直して置き直したファイルが黙って捨てられます。
名前に元オブジェクトのeTagを混ぜると、再配信による二重ロードを防ぎつつ、直したファイルは別名として扱われます。
一方で、このシリーズの出発点だったハードルの高さは変わっていません。
ユーザーはAWSにログインしてS3にファイルを置く必要があり、今回増えたのはメールが返ってくることだけです。
次回こそは、ユーザーがAWSを触らずに済む方法として、Streamlit in Snowflakeを使った経路を試します。
記事3行概要(再掲)
- Snowpipeの手前にLambdaを1つ置くと、壊れたCSVをテーブルに入れる前に弾いて、理由をメールでユーザーに差し戻せるよ！
- 検証OKのファイルは、名前を変えてからSnowpipeに渡さないといけないよ！Snowpipeは同じ名前のファイルを、中身が変わっていても再ロードしないよ！
- 名前に混ぜるのはタイムスタンプではなく元ファイルのeTagだよ！同じファイルでLambdaが2回動いても二重ロードにならないよ！