オプション1:Snowpipe REST API を使用したデータのロード

このトピックでは、パブリック REST エンドポイントを呼び出してデータをロードし、ロード履歴レポートを取得する方法について説明します。手順は、 Snowpipe REST APIを使用したデータロードの準備 の設定手順を完了していることを前提としています。

このトピックの内容:

データのロード

2つのステップでロードします。

ステップ1

データファイルをステージングします。

  • 内部ステージ: PUT コマンドを使用して、ファイルをステージングします。

  • 外部ステージ:クラウドプロバイダーが提供するクライアントツールを使用して、ファイルをステージの場所(Amazon S3、Google Cloud Storage、またはMicrosoft Azure)にコピーします。

ステップ2

insertFiles RESTエンドポイントにリクエストを送信して、ステージングされたデータファイルをロードします。

便宜上、このトピックでは、 REST エンドポイントを送信する方法を示すサンプルのJavaおよびPythonプログラムを提供しています。

Java SDK のサンプルプログラム

import net.snowflake.ingest.SimpleIngestManager;
import net.snowflake.ingest.connection.HistoryRangeResponse;
import net.snowflake.ingest.connection.HistoryResponse;
import org.bouncycastle.asn1.pkcs.PrivateKeyInfo;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.openssl.PEMParser;
import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
import org.bouncycastle.openssl.jcajce.JceOpenSSLPKCS8DecryptorProviderBuilder;
import org.bouncycastle.operator.InputDecryptorProvider;
import org.bouncycastle.operator.OperatorCreationException;
import org.bouncycastle.pkcs.PKCS8EncryptedPrivateKeyInfo;
import org.bouncycastle.pkcs.PKCSException;
import java.io.FileReader;
import java.io.IOException;
import java.nio.file.Paths;
import java.security.PrivateKey;
import java.security.Security;
import java.time.Instant;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class SDKTest
{
  // Path to the private key file that you generated earlier.
  private static final String PRIVATE_KEY_FILE = "/<path>/rsa_key.p8";

  public static class PrivateKeyReader
  {
    // If you generated an encrypted private key, implement this method to return
    // the passphrase for decrypting your private key.
    private static String getPrivateKeyPassphrase() {
      return "<private_key_passphrase>";
    }

    public static PrivateKey get(String filename)
            throws Exception
    {
      PrivateKeyInfo privateKeyInfo = null;
      Security.addProvider(new BouncyCastleProvider());
      // Read an object from the private key file.
      PEMParser pemParser = new PEMParser(new FileReader(Paths.get(filename).toFile()));
      Object pemObject = pemParser.readObject();
      if (pemObject instanceof PKCS8EncryptedPrivateKeyInfo) {
        // Handle the case where the private key is encrypted.
        PKCS8EncryptedPrivateKeyInfo encryptedPrivateKeyInfo = (PKCS8EncryptedPrivateKeyInfo) pemObject;
        String passphrase = getPrivateKeyPassphrase();
        privateKeyInfo = encryptedPrivateKeyInfo.decryptPrivateKeyInfo(pkcs8Prov);
      } else if (pemObject instanceof PrivateKeyInfo) {
        // Handle the case where the private key is unencrypted.
        privateKeyInfo = (PrivateKeyInfo) pemObject;
      }
      pemParser.close();
      JcaPEMKeyConverter converter = new JcaPEMKeyConverter().setProvider(BouncyCastleProvider.PROVIDER_NAME);
      return converter.getPrivateKey(privateKeyInfo);
    }
  }

  private static HistoryResponse waitForFilesHistory(SimpleIngestManager manager,
                                                     Set<String> files)
          throws Exception
  {
    ExecutorService service = Executors.newSingleThreadExecutor();

    class GetHistory implements
            Callable<HistoryResponse>
    {
      private Set<String> filesWatchList;
      GetHistory(Set<String> files)
      {
        this.filesWatchList = files;
      }
      String beginMark = null;

      public HistoryResponse call()
              throws Exception
      {
        HistoryResponse filesHistory = null;
        while (true)
        {
          Thread.sleep(500);
          HistoryResponse response = manager.getHistory(null, null, beginMark);
          if (response.getNextBeginMark() != null)
          {
            beginMark = response.getNextBeginMark();
          }
          if (response != null && response.files != null)
          {
            for (HistoryResponse.FileEntry entry : response.files)
            {
              //if we have a complete file that we've
              // loaded with the same name..
              String filename = entry.getPath();
              if (entry.getPath() != null && entry.isComplete() &&
                      filesWatchList.contains(filename))
              {
                if (filesHistory == null)
                {
                  filesHistory = new HistoryResponse();
                  filesHistory.setPipe(response.getPipe());
                }
                filesHistory.files.add(entry);
                filesWatchList.remove(filename);
                //we can return true!
                if (filesWatchList.isEmpty()) {
                  return filesHistory;
                }
              }
            }
          }
        }
      }
    }

    GetHistory historyCaller = new GetHistory(files);
    //fork off waiting for a load to the service
    Future<HistoryResponse> result = service.submit(historyCaller);

    HistoryResponse response = result.get(2, TimeUnit.MINUTES);
    return response;
  }

  public static void main(String[] args)
  {
    final String host = "<account_name>.<region_id>.snowflakecomputing.com";
    final String user = "<user_login_name>";
    final String pipe = "<db_name>.<schema_name>.<pipe_name>";
    try
    {
      final long oneHourMillis = 1000 * 3600L;
      String startTime = Instant
              .ofEpochMilli(System.currentTimeMillis() - 4 * oneHourMillis).toString();
      final PrivateKey privateKey = PrivateKeyReader.get(PRIVATE_KEY_FILE);
      SimpleIngestManager manager = new SimpleIngestManager(host.split("\.")[0], user, pipe, privateKey, "https", host, 443);
      Set<String> files = new TreeSet<>();
      // Add the paths to the files that you want to load.
      // Use paths that are relative to the stage where the files are located
      // (the stage that is specified in the pipe definition)..
      files.add("<path>/<filename>");
      files.add("<path>/<filemame>");
      ...
      manager.ingestFiles(manager.wrapFilepaths(files), null);
      HistoryResponse history = waitForFilesHistory(manager, files);
      System.out.println("Received history response: " + history.toString());
      String endTime = Instant
              .ofEpochMilli(System.currentTimeMillis()).toString();

      HistoryRangeResponse historyRangeResponse =
              manager.getHistoryRange(null,
                                      startTime,
                                      endTime);
      System.out.println("Received history range response: " +
                                 historyRangeResponse.toString());

    }
    catch (Exception e)
    {
      e.printStackTrace();
    }

  }
}

この例では、 Bouncy Castle暗号化 APIs を使用しています。この例をコンパイルして実行するには、クラスパスに次の JAR ファイルを含める必要があります。

  • プロバイダー JAR ファイル( bcprov-jdkバージョン.jar

  • PKIX / CMS / EAC / PKCS / OCSP / TSP / OPENSSL JAR ファイル( bcpkix-jdkバージョン.jar

ここで、 バージョン は、 JAR ファイルがサポートする JDK のバージョンを指定します。

サンプルコードをコンパイルする前に、次のプレースホルダー値を置き換えます。

PRIVATE_KEY_FILE = "/<パス>/rsa_key.p8

キーペア認証の使用およびキーローテーションSnowpipe REST APIを使用したデータロードの準備 内)で作成した秘密キーファイルへのローカルパスを指定します。

getPrivateKeyPassphrase() 内の return "<秘密キーパスフレーズ>"

暗号化されたキーを生成した場合は、 getPrivateKeyPassphrase() メソッドを実装して、そのキーを復号化するためのパスフレーズを返します。

host = "<アカウント名>.<地域ID>.snowflakecomputing.com"

ホスト情報を URL ドメインの形式で指定します。 URL の形式は、アカウントが置かれている 地域 によって異なります。

US 西部

<アカウント名>.snowflakecomputing.com

その他の地域

<アカウント名>.<地域ID>.snowflakecomputing.com

<アカウント名> はアカウントの名前(Snowflakeが提供)で、 <地域ID> は次のとおりです。

地域

地域 ID

メモ

Amazon Web Services(AWS)

US 西部(オレゴン)

us-west-2

US 西部のアカウントに AWS PrivateLink を設定する場合にのみ必要です。

US 東部(オハイオ)

us-east-2.aws

US 東部(バージニア北部)

us-east-1

US 東部(商業組織、バージニア政府北部)

us-east-1-gov.aws

Business Critical(またはそれ以上)のアカウントでのみ使用できます。Snowflakeがまだサポートしていない独立した専用クラウドである、 AWS GovCloud (US)には位置していません。

カナダ(中部)

ca-central-1.aws

EU (アイルランド)

eu-west-1

ヨーロッパ(ロンドン)

eu-west-2.aws

EU (フランクフルト)

eu-central-1

アジア太平洋(東京)

ap-northeast-1.aws

アジア太平洋(ムンバイ)

ap-south-1.aws

アジア太平洋(シンガポール)

ap-southeast-1

アジア太平洋(シドニー)

ap-southeast-2

Google Cloud Platform(GCP)

US 中央部1(アイオワ)

us-central1.gcp

ヨーロッパ西部2(ロンドン)

europe-west2.gcp

ヨーロッパ西部4(オランダ)

europe-west4.gcp

Microsoft Azure

西 US 2(ワシントン)

west-us-2.azure

東 US 2(バージニア)

east-us-2.azure

US 政府バージニア

us-gov-virginia.azure

Business Critical(またはそれ以上)のアカウントでのみ利用可能です。

カナダ中央部(トロント)

canada-central.azure

西ヨーロッパ(オランダ)

west-europe.azure

スイス北部(チューリッヒ)

switzerland-north.azure

東南アジア(シンガポール)

southeast-asia.azure

オーストラリア東部(ニューサウスウェールズ)

australia-east.azure

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

Snowflakeログイン名を指定します。

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

データのロードに使用するパイプの完全修飾名を指定します。

files.add("<パス>/<ファイル名>")

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

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

Python SDK のサンプルプログラム

from logging import getLogger
from snowflake.ingest import SimpleIngestManager
from snowflake.ingest import StagedFile
from snowflake.ingest.utils.uris import DEFAULT_SCHEME
from datetime import timedelta
from requests import HTTPError
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.hazmat.primitives.serialization import PrivateFormat
from cryptography.hazmat.primitives.serialization import NoEncryption
import time
import datetime
import os
import logging

logging.basicConfig(
        filename='/tmp/ingest.log',
        level=logging.DEBUG)
logger = getLogger(__name__)

# If you generated an encrypted private key, implement this method to return
# the passphrase for decrypting your private key.
def get_private_key_passphrase():
  return '<private_key_passphrase>'

with open("/<private_key_path>/rsa_key.p8", 'rb') as pem_in:
  pemlines = pem_in.read()
  private_key_obj = load_pem_private_key(pemlines,
  get_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
file_list=['<path>/<filename>', '<path>/<filename>']
ingest_manager = SimpleIngestManager(account='<account_name>',
                                     host='<account_name>.<region_id>.snowflakecomputing.com',
                                     user='<user_login_name>',
                                     pipe='<db_name>.<schema_name>.<pipe_name>',
                                     private_key=private_key_text)
# List of files, but wrapped into a class
staged_file_list = []
for file_name in file_list:
    staged_file_list.append(StagedFile(file_name, None))

try:
    resp = ingest_manager.ingest_files(staged_file_list)
except HTTPError as e:
    # HTTP error, may need to retry
    logger.error(e)
    exit(1)

# This means Snowflake has received file and will start loading
assert(resp['responseCode'] == 'SUCCESS')

# Needs to wait for a while to get result in history
while True:
    history_resp = ingest_manager.get_history()

    if len(history_resp['files']) > 0:
        print('Ingest Report:\n')
        print(history_resp)
        break
    else:
        # wait for 20 seconds
        time.sleep(20)

    hour = timedelta(hours=1)
    date = datetime.datetime.utcnow() - hour
    history_range_resp = ingest_manager.get_history_range(date.isoformat() + 'Z')

    print('\nHistory scan report: \n')
    print(history_range_resp)

サンプルコードを実行する前に、次のプレースホルダー値を置き換えます。

<秘密キーパス>

キーペア認証の使用およびキーローテーションSnowpipe REST APIを使用したデータロードの準備 内)で作成した秘密キーファイルへのローカルパスを指定します。

get_private_key_passphrase() 内の return "<秘密キーパスフレーズ>"

暗号化されたキーを生成した場合は、 get_private_key_passphrase() 関数を実装して、そのキーを復号化するためのパスフレーズを返します。

account='<アカウント名>'

アカウントの名前を指定します(Snowflakeが提供)。

host='<アカウント名>.<地域ID>.snowflakecomputing.com'

ホスト情報を URL ドメインの形式で指定します。 URL ドメインの形式は、アカウントが置かれている 地域 によって異なります。

US 西部

<アカウント名>.snowflakecomputing.com

その他の地域

<アカウント名>.<地域ID>.snowflakecomputing.com

<アカウント名> はアカウントの名前(Snowflakeが提供)で、 <region_id> は次のとおりです。

地域

地域 ID

メモ

Amazon Web Services(AWS)

US 西部(オレゴン)

us-west-2

US 西部のアカウントに AWS PrivateLink を設定する場合にのみ必要です。

US 東部(オハイオ)

us-east-2.aws

US 東部(バージニア北部)

us-east-1

US 東部(商業組織、バージニア政府北部)

us-east-1-gov.aws

Business Critical(またはそれ以上)のアカウントでのみ使用できます。Snowflakeがまだサポートしていない独立した専用クラウドである、 AWS GovCloud (US)には位置していません。

カナダ(中部)

ca-central-1.aws

EU (アイルランド)

eu-west-1

ヨーロッパ(ロンドン)

eu-west-2.aws

EU (フランクフルト)

eu-central-1

アジア太平洋(東京)

ap-northeast-1.aws

アジア太平洋(ムンバイ)

ap-south-1.aws

アジア太平洋(シンガポール)

ap-southeast-1

アジア太平洋(シドニー)

ap-southeast-2

Google Cloud Platform(GCP)

US 中央部1(アイオワ)

us-central1.gcp

ヨーロッパ西部2(ロンドン)

europe-west2.gcp

ヨーロッパ西部4(オランダ)

europe-west4.gcp

Microsoft Azure

西 US 2(ワシントン)

west-us-2.azure

東 US 2(バージニア)

east-us-2.azure

US 政府バージニア

us-gov-virginia.azure

Business Critical(またはそれ以上)のアカウントでのみ利用可能です。

カナダ中央部(トロント)

canada-central.azure

西ヨーロッパ(オランダ)

west-europe.azure

スイス北部(チューリッヒ)

switzerland-north.azure

東南アジア(シンガポール)

southeast-asia.azure

オーストラリア東部(ニューサウスウェールズ)

australia-east.azure

user='<user_login_name>'

Snowflakeログイン名を指定します。

pipe='<db_name>.<schema_name>.<pipe_name>'

データのロードに使用するパイプの完全修飾名を指定します。

file_list=['<path>/<filename>', '<path>/<filename>']

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

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

ロード履歴を表示する

Snowflakeは、ロード履歴を表示するための REST エンドポイント情報スキーマ テーブル関数を提供します。

REST エンドポイントの呼び出しとは異なり、情報スキーマテーブル関数またはアカウント使用状況ビューのいずれかをクエリするには、実行中のウェアハウスが必要です。