Option 2: Automate Snowpipe with AWS Lambda¶

ステップ1: Snowpipe REST API を呼び出すPythonコードを書く¶

サンプルPythonコード

from __future__ import print_function
from snowflake.ingest import SimpleIngestManager
from snowflake.ingest import StagedFile
from requests import HTTPError
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.hazmat.primitives.serialization import PrivateFormat
from cryptography.hazmat.primitives.serialization import NoEncryption
from cryptography.hazmat.backends import default_backend

import os

with open("./rsa_key.p8", 'rb') as pem_in:
  pemlines = pem_in.read()
  private_key_obj = load_pem_private_key(pemlines,
  os.environ['PRIVATE_KEY_PASSPHRASE'].encode(),
  default_backend())

private_key_text = private_key_obj.private_bytes(
  Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()).decode('utf-8')
# Assume the public key has been registered in Snowflake:
# private key in PEM format

# List of files in the stage specified in the pipe definition
ingest_manager = SimpleIngestManager(account='<account_identifier>',
                   host='<account_identifier>.snowflakecomputing.com',
                   user='<user_login_name>',
                   pipe='<db_name>.<schema_name>.<pipe_name>',
                   private_key=private_key_text)

def handler(event, context):
  for record in event['Records']:
    bucket = record['s3']['bucket']['name']
    key = record['s3']['object']['key']

    print("Bucket: " + bucket + " Key: " + key)
    # List of files in the stage specified in the pipe definition
    # wrapped into a class
    staged_file_list = []
    staged_file_list.append(StagedFile(key, None))

    print('Pushing file list to ingest REST API')
    resp = ingest_manager.ingest_files(staged_file_list)

サンプルコードを使用する前に、次を変更します。

1. セキュリティパラメーターを更新します。

   private_key=""" / -----BEGIN RSA PRIVATE KEY----- / ... / -----END RSA PRIVATE KEY----- """

   Use key pair authentication & key rotation （ Data loading preparation using the Snowpipe REST API 内）で作成した秘密キーファイルのコンテンツを指定します。

   PRIVATE_KEY_PASSPHRASE 環境変数を使用して、秘密キーファイルを復号化するためのパスフレーズを指定します。

   • Linuxまたは macOS：

     CopyExpand

     export PRIVATE_KEY_PASSPHRASE='<passphrase>'
     

     Show lessSee more

     Scroll to top

   • Windows：

     CopyExpand

     set PRIVATE_KEY_PASSPHRASE='<passphrase>'
     

     Show lessSee more

     Scroll to top

2. セッションパラメーターを更新します。

   account='<アカウント識別子>'

   アカウントの一意の識別子を指定します（Snowflakeが提供）。 host の説明もご参照ください。

   host='<アカウント識別子>.snowflakecomputing.com'

   Snowflakeアカウントの一意のホスト名を指定します。

   アカウント識別子の推奨形式は次のとおりです。

   organization_name-account_name

   Snowflake組織とアカウントの名前。詳細については、 形式1（推奨）: 組織内のアカウント名 をご参照ください。

   または、必要に応じて、アカウントがホストされている リージョン および クラウドプラットフォーム とともに、 アカウントロケーター を指定します。詳細については、 形式2: リージョン内のアカウントロケーター をご参照ください。

   user='<ユーザーログイン名>'

   Snowpipeコードを実行するSnowflakeユーザーのログイン名を指定します。

   pipe='<データベース名>.<スキーマ名>.<パイプ名>'

   データのロードに使用するパイプの完全修飾名を <データベース名>.<スキーマ名>.<パイプ名> の形式で指定します。

3. ファイルオブジェクトリストでインポートするファイルへのパスを指定します。

   staged_file_list = []

   指定するパスは、ファイルが配置されているステージを 基準 にする必要があります。ファイル拡張子を含む各ファイルの完全な名前を含めます。たとえば、gzip圧縮された CSV ファイルには、拡張子 .csv.gz が付いている場合があります。

4. ファイルを便利な場所に保存します。

このトピックの残りの手順では、ファイル名を SnowpipeLambdaCode.py と想定しています。

ステップ2: Lambda関数デプロイメントパッケージを作成する¶

次の手順を実行してLambda用のPythonランタイム環境を構築し、 ステップ1： Snowpipe REST API を呼び出すPythonコードを書く （このトピック内）に合わせたSnowpipeのコードを追加します。これらの手順の詳細については、 AWS Lambdaデプロイメントパッケージドキュメント をご参照ください（Pythonの手順を参照）。

1. AWS EC2 の手順 を完了して、AWS EC2 Linuxインスタンスを作成します。このインスタンスは、Snowpipeコードを実行するためのコンピューティングリソースを提供します。

2. SCP （セキュアコピー）を使用して、Snowpipeコードファイルを新しい AWS EC2 インスタンスにコピーします。

   CopyExpand

   scp -i key.pem /<path>/SnowpipeLambdaCode.py ec2-user@<machine>.<region_id>.compute.amazonaws.com:~/SnowpipeLambdaCode.py
   

   Show lessSee more

   Scroll to top

   条件：

   • <パス> は、ローカルの SnowpipeLambdaCode.py ファイルへのパスです。

   • <マシン>.<地域ID> は、 EC2 インスタンスの DNS 名です（例： ec2-54-244-54-199.us-west-2.compute.amazonaws.com）。

     DNS 名は、Amazon EC2 コンソールの Instances 画面に表示されます。

3. SSH を使用してEC2 インスタンスに接続します（セキュア SHell）。

   CopyExpand

   ssh -i key.pem ec2-user@<machine>.<region_id>.compute.amazonaws.com
   

   Show lessSee more

   Scroll to top

4. Pythonおよび関連ライブラリを EC2 インスタンスにインストールします。

   CopyExpand

   sudo yum install -y gcc zlib zlib-devel openssl openssl-devel
   
   wget https://www.python.org/ftp/python/3.6.1/Python-3.6.1.tgz
   
   tar -xzvf Python-3.6.1.tgz
   
   cd Python-3.6.1 && ./configure && make
   
   sudo make install
   
   sudo /usr/local/bin/pip3 install virtualenv
   
   /usr/local/bin/virtualenv ~/shrink_venv
   
   source ~/shrink_venv/bin/activate
   
   pip install Pillow
   
   pip install boto3
   
   pip install requests
   
   pip install snowflake-ingest
   

   Show lessSee more

   Scroll to top

5. .zipデプロイメントパッケージを作成します（Snowpipe.zip）。

   CopyExpand

   cd $VIRTUAL_ENV/lib/python3.6/site-packages
   
   zip -r9 ~/Snowpipe.zip .
   
   cd ~
   
   zip -g Snowpipe.zip SnowpipeLambdaCode.py
   

   Show lessSee more

   Scroll to top

ステップ3: Lambdaの AWS IAM ロールを作成する¶

AWS Lambdaドキュメント に従って、Lambda関数を実行する IAM ロールを作成します。

ロールの IAM Amazonリソース名（ARN） を記録します。次のステップで使用します。

ステップ4: Lambda関数を作成する¶

ステップ2：Lambda関数デプロイメントパッケージを作成する （このトピック内）で作成した、 .zip デプロイメントパッケージをアップロードしてLambda関数を作成します。

CopyExpand

aws lambda create-function \
--region us-west-2 \
--function-name IngestFile \
--zip-file fileb://~/Snowpipe.zip \
--role arn:aws:iam::<aws_account_id>:role/lambda-s3-execution-role \
--handler SnowpipeLambdaCode.handler \
--runtime python3.6 \
--profile adminuser \
--timeout 10 \
--memory-size 1024

Show lessSee more

Scroll to top

--role には、 ステップ3：Lambdaの AWS IAM ロールを作成する （このトピック内）で記録したロール ARN を指定します。

出力から新しい関数の ARN を記録します。次のステップで使用します。

ステップ5: Lambda関数の呼び出しを許可する¶

S3に関数を呼び出すために必要な権限を付与します。

--source-arn には、 ステップ4：Lambda関数を作成する （このトピック内）で記録した関数 ARN を指定します。

CopyExpand

aws lambda add-permission \
--function-name IngestFile \
--region us-west-2 \
--statement-id enable-ingest-calls \
--action "lambda:InvokeFunction" \
--principal s3.amazonaws.com \
--source-arn arn:aws:s3:::<SourceBucket> \
--source-account <aws_account_id> \
--profile adminuser

Show lessSee more

Scroll to top

ステップ6: Lambda通知イベントを登録する¶

Amazon S3イベント通知 の手順を完了して、Lambda通知イベントを登録します。入力フィールドで、 ステップ4：Lambda関数を作成する （このトピック内）で記録した関数 ARN を指定します。