オプション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();
        InputDecryptorProvider pkcs8Prov = new JceOpenSSLPKCS8DecryptorProviderBuilder().build(passphrase.toCharArray());
        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_identifier>.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);
      List<StagedFileWrapper> files = new ArrayList<>();
      // Add the paths and sizes 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(new StagedFileWrapper("<path>/<filename>", <file_size_in_bytes> /* file size is optional but recommended, pass null when it is not available */));
      files.add(new StagedFileWrapper("<path>/<filename>", <file_size_in_bytes> /* file size is optional but recommended, pass null when it is not available */));
      ...
      manager.ingestFiles(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();
    }

  }
}
Copy

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

  • プロバイダー JAR ファイル( bcprov-jdkversions.jar

  • PKIX / CMS / EAC / PKCS / OCSP / TSP / OPENSSL JAR ファイル( bcpkix-jdkversions.jar

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

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

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

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

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

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

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

ホスト情報を URL の形式で指定します。

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

organization_name-account_name

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

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

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

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

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

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

files.add("<パス>/<ファイル名>", <バイト単位のファイルサイズ>)

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

オプションで、Snowpipeがデータのロードに必要な操作を計算するときの遅延を回避するために、各ファイルのサイズをバイト単位で指定します。

指定するパスは、ファイルが配置されているステージを 基準 にする必要があります。ファイル拡張子を含む各ファイルの完全な名前を含めます。たとえば、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

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)
# List of files, but wrapped into a class
staged_file_list = [
  StagedFile('<path>/<filename>', <file_size_in_bytes>),  # file size is optional but recommended, pass None if not available
  StagedFile('<path>/<filename>', <file_size_in_bytes>),  # file size is optional but recommended, pass None if not available
  ...
  ]

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)
Copy

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

<秘密キーパス>

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

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

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

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

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

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

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

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

organization_name-account_name

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

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

user='<user_login_name>'

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

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

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

file_list=['<path>/<filename>', '<path>/<filename>'] | staged_file_list=[StagedFile('<path>/<filename>', <file_size_in_bytes>), StagedFile('<path>/<filename>', <file_size_in_bytes>)]

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

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

オプションで、Snowpipeがデータのロードに必要な操作を計算するときの遅延を回避するために、各ファイルのサイズをバイト単位で指定します。

ロード履歴を表示する

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

REST エンドポイントの呼び出しとは異なり、Information Schemaテーブル関数またはAccount Usageビューのいずれかをクエリするには、稼働中のウェアハウスが必要です。

ステージングされたファイルの削除

データが正常にロードされ、ファイルが不要になったら、ステージングされたファイルを削除します。手順については、 Snowpipeがデータをロードした後のステージングされたファイルの削除 をご参照ください。