SnowflakeのSQL APIをRubyで試してみた

こんにちは。クラシル開発部でバックエンドエンジニアの高松です（@takarotoooooo）

この記事は dely Advent Calendar 2021 12日目の記事です。

昨日はknchstさんの「クラシルiOSのパッケージマネジメントについて」というお話でしたiOSの開発にも興味がある方はぜひ見てみてください！

今回は先日データエンジニアのharryさんが「クラシルでのSnowflakeデータパイプラインのお話＆活用Tips」で紹介していたSnowflakeをアプリケーション側から利用する方法を試してみたので紹介します

はじめに

アプリケーションからログデータを利用できると何がうれしいのか？

クラシルでは現在Snowflakeを用いてログデータの可視化・分析を行っています
分析結果を元に意思決定を行い、改修することで収集したデータを開発に還元しています

しかし可視化・分析だけでなく、アプリケーションからログデータを利用できるようにすることで、コンテンツ配信やレコメンドの最適化など実現可能な施作を増やすことができるため、結果としてサービス、プロダクトで解決できる課題の幅を広げることができます

注意点

今回はキーペア認証方式で利用する方法をご紹介しています
また、SQL APIはプレビュー機能なので正式リリースされる際には仕様が少し変わっている可能性があるのでご注意ください
docs.snowflake.com

準備

公開鍵と秘密鍵のペアを生成
公開鍵をSnowflakeユーザーに割り当てる

公開鍵と秘密鍵のペアを生成

秘密鍵を生成します

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8

上で生成した秘密鍵を元に公開鍵を生成します

openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

公開鍵をSnowflakeユーザーに割り当てる

上で発行した公開鍵の内容を接続に利用するユーザーに設定します。
（ユーザーの変更はSECURITYADMINロールのユーザー以上である必要があるので、権限が足りない場合は然るべき人にお願いしてください）

alter user [USER_NAME] set rsa_public_key='[PUBLIC_KEY_VALUE]';

接続してみる

必要なもの

秘密鍵
Snowflakeのアカウント識別子
Snowflakeのアカウント
Snowflakeのユーザー名

公開鍵のフィンガープリントを作る

private_key = OpenSSL::PKey::RSA.new(pemlines)
public_key = private_key.public_key
public_key_fp = 
  %(SHA256:#{Base64.encode64(Digest::SHA256.digest(public_key.to_der)).strip})

JWTの作成

ペイロードに以下のフィールドを持つJWTを作成します
なお、ここではGemjwtを利用して生成を行なっています

フィールド	値
iss	`<アカウント識別子>`.`<ユーザー>`.`<公開鍵のフィンガープリント>`
sub	`<アカウント識別子>`.`<ユーザー>`
iat	JWT が発行された時間（UTC のエポック開始からの秒数）
exp	JWT の有効期限が切れる時間（UTC のエポック開始からの秒数）

jwt_created_at = Time.now.getutc
lifetime = 60 * 60
qualified_username = %(#{account.upcase}.#{user_name.upcase})
payload = {
  'iss': %(#{qualified_username}.#{public_key_fp}),
  'sub': qualified_username,
  'iat': jwt_created_at.to_i,
  'exp': (jwt_created_at + lifetime).to_i
}

algorithm = 'RS256'
token = JWT.encode(payload, private_key, algorithm)

APIリクエスト

生成したJWTを利用して、下記のURLにPOSTリクエストを送信します

https://<アカウント識別子>.snowflakecomputing.com/api/statements

リクエストヘッダ

項目名	値
Authorization	Bearer `< JWT >`
Accept	application/json
Content-Type	application/json
X-Snowflake-Authorization-Token-Type	KEYPAIR_JWT

リクエストボディ

項目名	値
statement	実行するSQL
timeout	タイムアウト秒数
database	データベース名
schema	スキーマ名
warehouse	ウェアハウス名
role	ロール

リクエストパラメータ

項目名	値
requestId	リクエスト毎のuuid
async	falseで同期的にデータを取得
pageSize	取得するデータサイズ

http = Net::HTTP.new(%(#{<アカウント識別子>}.snowflakecomputing.com), 443)
http.use_ssl = true

query_string = {
  requestId: SecureRandom.uuid,
  async: false,
  pageSize: 15,
  nullable: true
}.map { |k, v| %(#{k}=#{v}) }.join('&')
path = %(#{uri.path}?#{query_string})

header = {
  'Authorization': %(Bearer #{token}),
  'Accept': 'application/json',
  'Content-Type': 'application/json',
  'User-Agent': 'applicationName/applicationVersion',
  'X-Snowflake-Authorization-Token-Type': 'KEYPAIR_JWT'
}

params = {
  statement:'SELECT * FROM CUSTOMER;',
  timeout: 60,
  database: 'SNOWFLAKE_SAMPLE_DATA',
  schema: 'TPCH_SF1',
  warehouse:  <ウェアハウス名>,
  role: <ロール>,
}
response = http.post(path, params.to_json, header)

pp JSON.parse(response)

こんな感じで結果が返ってきます

{"resultSetMetaData"=>
  {"page"=>0,
   "pageSize"=>15,
   "numPages"=>10000,
   "numRows"=>150000,
   "format"=>"json",
   "rowType"=>
    [{"name"=>"C_CUSTKEY",
      "database"=>"SNOWFLAKE_SAMPLE_DATA",
      "schema"=>"TPCH_SF1",
      "table"=>"CUSTOMER",
      "scale"=>0,
      "precision"=>38,
~~~~~
"data"=>
  [["0",
    "30001",
    "Customer#000030001",
    "Ui1b,3Q71CiLTJn4MbVp,,YCZARIaNTelfst",
    "4",
    "14-526-204-4500",
    "8848.47",
    "MACHINERY",
    "frays wake blithely enticingly ironic asymptote"],
~~~~~
"code"=>"090001",
"statementStatusUrl"=>
  "/api/statements/01a0cf10-0000-30b4-0000-40d5007afd72?requestId=30139506-efcb-40c0-af02-1cfb97c64b30&pageSize=15",
 "requestId"=>"30139506-efcb-40c0-af02-1cfb97c64b30",
 "sqlState"=>"00000",
 "statementHandle"=>"01a0d3bf-0000-30ee-0000-40d5007cf11e",
 "message"=>"Statement executed successfully.",
 "createdOn"=>1639031029314}