#15 ： Golang用のSnowflake Driverを使ってSnowflakeに接続＆クエリを実行する – Snowflake Advent Calendar 2019

#15 ： Golang用のSnowflake Driverを使ってSnowflakeに接続＆クエリを実行する – Snowflake Advent Calendar 2019

「カスタマーストーリー」を支えるツール特集

「カスタマーストーリー」を支えるツール特集

#ビッグデータ
#ビジネス・アナリティクス
#Snowflake
#Go
八幡豊

八幡豊

facebook logohatena logotwitter logo
Clock Icon2019.12.15

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

本エントリは『Snowflake Advent Calendar 2019』の15日目のエントリです。

Snowflakeには各種プログラミング言語向けのコネクターやドライバーがあります。現時点でSnowflake公式のものとしては以下の8つが利用可能です。

  • Snowflake Connector for Python
  • Snowflake Connector for Spark
  • Snowflake Connector for Kafka
  • Node.js Driver
  • Go Snowflake Driver(gosnowflake)
  • .NET Driver
  • JDBC Driver
  • ODBC Driver

本ブログ記事では、このうちGolang用のパッケージGo Snowflake Driver(gosnowflake)を使ってSnowflakeに接続する方法をご紹介します。

動作環境

  • OS : macOS Catalina (v10.15.2)
  • Golang : v1.13.4
  • Go Snowflake Driver : v1.3.3

本ブログ記事中のサンプルプログラムでは、「Snowflake in 20 Minutes」で作成したデータベースに接続＆クエリを実行しています。実際にサンプルプログラムを動かす場合は、以下のブログ記事を参考にサンプルデータベースを作成ください。

gosnowflakeについて

gosnowflakeはdatabase/sqlパッケージ用のドライバーです。データベースへの接続やクエリ実行方法はdatabase/sqlの作法に則ります。

プロジェクトの作成

プロジェクトディレクトリを作成し、go mod initで初期化します。

$ mkdir gosnowflake-sample
$ cd gosnowflake-sample
$ go mod init github.com/yuyawata/gosnowflake-sample

パスワード認証で接続

まずはユーザー名とパスワードを指定してSnowflakeに接続＆クエリを実行してみます。

サンプルプログラム

package main

import (
    "database/sql"
    "flag"
    "fmt"
    "log"
    "os"

    "github.com/olekukonko/tablewriter"

    "github.com/aws/aws-sdk-go/aws"

    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/ssm"
    ssmconfig "github.com/ianlopshire/go-ssm-config"

    sf "github.com/snowflakedb/gosnowflake"
)

type Config struct {
    Account   string `ssm:"Account" required:"true"`
    User      string `ssm:"User" required:"true"`
    Password  string `ssm:"Password" required:"true"`
    Database  string `ssm:"Database"`
    Warehouse string `ssm:"Warehouse"`
}

func getDSN() (string, *sf.Config, error) {

    sess := session.Must(session.NewSession())

    provider := &ssmconfig.Provider{
        SSM: ssm.New(sess, aws.NewConfig().WithRegion("ap-northeast-1")),
    }

    var c Config
    err := provider.Process("/Snowflake/Dev/", &c)
    if err != nil {
        return "", nil, err
    }

    cfg := &sf.Config{
        Account:   c.Account,
        User:      c.User,
        Password:  c.Password,
        Database:  c.Database,
        Warehouse: c.Warehouse,
    }

    dsn, err := sf.DSN(cfg)
    return dsn, cfg, err
}

func run(dsn string) error {
    db, err := sql.Open("snowflake", dsn)
    if err != nil {
        log.Fatalf("failed to connect. %v, err: %v", dsn, err)
        return err
    }
    defer db.Close()

    query := "select email from emp_basic where email like '%.uk';"
    rows, err := db.Query(query)
    if err != nil {
        log.Fatalf("failed to run a query. %v, err: %v", query, err)
        return err
    }
    defer rows.Close()

    table := tablewriter.NewWriter(os.Stdout)
    table.SetHeader([]string{"Email"})
    var email string

    for rows.Next() {
        err := rows.Scan(&email)
        if err != nil {
            log.Fatalf("failed to get result. err: %v", err)
        }
        table.Append([]string{email})
    }
    if rows.Err() != nil {
        fmt.Printf("ERROR: %v\n", rows.Err())
        return rows.Err()
    }
    table.Render()
    return nil
}

func main() {
    if !flag.Parsed() {
        flag.Parse()
    }

    dsn, cfg, err := getDSN()
    if err != nil {
        log.Fatalf("failed to create DSN from Config: %v, err: %v", cfg, err)
        os.Exit(1)
    }

    err = run(dsn)
    if err != nil {
        log.Fatalf("failed to execute a query, err: %v", err)
        os.Exit(1)
    }
}

ビルド

go buildを実行すると、、、以下のようなエラーが発生します。

$ go build
go: finding github.com/aws/aws-sdk-go v1.26.2
go: downloading github.com/aws/aws-sdk-go v1.26.2
go: extracting github.com/aws/aws-sdk-go v1.26.2
go: downloading golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2
go: finding github.com/pkg/browser latest
go: extracting golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2
# github.com/snowflakedb/gosnowflake
../../../../pkg/mod/github.com/snowflakedb/gosnowflake@v1.3.3/auth.go:393:8: claims.SetIssuer undefined (type jws.Claims has no field or method SetIssuer)
../../../../pkg/mod/github.com/snowflakedb/gosnowflake@v1.3.3/auth.go:395:8: claims.SetSubject undefined (type jws.Claims has no field or method SetSubject)
../../../../pkg/mod/github.com/snowflakedb/gosnowflake@v1.3.3/auth.go:396:8: claims.SetIssuedAt undefined (type jws.Claims has no field or method SetIssuedAt)
../../../../pkg/mod/github.com/snowflakedb/gosnowflake@v1.3.3/auth.go:397:8: claims.SetExpiration undefined (type jws.Claims has no field or method SetExpiration)`

これはgosnowflakeの依存パッケージJOSEがSemantic Versioningに対応していないために発生するエラーです。（Go ModulesでJOSEの最新のバージョンを正しく認識できず、v0.9.1が最新バージョンとして認識されてしまいます）。

対策として、replaceを使って手動で最新バージョンを指定します。具体的にはgo.modに以下を追加します。

replace github.com/SermoDigital/jose => github.com/SermoDigital/jose 803625baeddc3526d01d321b5066029f53eafc81

また、ロギング用のパッケージについてはsnowflakeの独自実装版のglogの利用が推奨されているので、こちらもreplaceでパッケージを置き換えます。

replace github.com/golang/glog => github.com/snowflakedb/glog f5055e6f21ce71153a199e6460664045b3af8e50

この時点でgo.modは以下のようになります。

module github.com/yuyawata/gosnowflake-sample

go 1.13

require (
    github.com/SermoDigital/jose v0.9.1 // indirect
    github.com/aws/aws-sdk-go v1.26.2
    github.com/google/uuid v1.1.1 // indirect
    github.com/ianlopshire/go-ssm-config v1.0.1
    github.com/olekukonko/tablewriter v0.0.4
    github.com/pkg/browser v0.0.0-20180916011732-0a3d74bf9ce4 // indirect
    github.com/snowflakedb/gosnowflake v1.3.3
)

replace github.com/SermoDigital/jose => github.com/SermoDigital/jose 803625baeddc3526d01d321b5066029f53eafc81
replace github.com/golang/glog => github.com/snowflakedb/glog f5055e6f21ce71153a199e6460664045b3af8e50

再度go buildを実行します。

$ go build
go: finding github.com/SermoDigital/jose 803625baeddc3526d01d321b5066029f53eafc81
go: finding github.com/snowflakedb/glog f5055e6f21ce71153a199e6460664045b3af8e50

今度はビルドに成功します。またgo.modのreplaceでのバージョン指定部分は以下のように自動で書き換わります。

replace github.com/SermoDigital/jose => github.com/SermoDigital/jose 803625baeddc3526d01d321b5066029f53eafc81
replace github.com/golang/glog => github.com/snowflakedb/glog f5055e6f21ce71153a199e6460664045b3af8e50
↓
replace github.com/SermoDigital/jose => github.com/SermoDigital/jose v0.9.2-0.20180104203859-803625baeddc
replace github.com/golang/glog => github.com/snowflakedb/glog v0.0.0-20180824191149-f5055e6f21ce

実行

クエリの実行に成功すると、以下のような結果が表示されます。

./gosnowflake-sample
+--------------------------+
|          EMAIL           |
+--------------------------+
| rtalmadgej@sf_tuts.co.uk |
| gbassfordo@sf_tuts.co.uk |
+--------------------------+

サンプルプログラムの解説

gosnowflakeでは接続文字列として以下のシンタックスが利用できます。

  • username[:password]@accountname/dbname/schemaname[?param1=value&...&paramN=valueN
  • username[:password]@accountname/dbname[?param1=value&...&paramN=valueN
  • username[:password]@hostname:port/dbname/schemaname?account=<your_account>[&param1=value&...&paramN=valueN]

接続文字列を生成するためのConfig構造体とDSNメソッドも用意されています。上記のサンプルプログラムではこれらを利用して接続文字列を生成しています。接続パラメーターの詳細についてはgosnowflakeのドキュメントを参照ください。

また上記のサンプルプログラムでは接続パラメーターをAWS Systems Manager Parameter Store（以降Parameter Store）から取得する形にしています。事前に、Parameter Storeに接続パラメーターを保存しておいてください。

例として、AWS CLIでParameter Storeにパラメーターを保存する方法を記載します。

## リージョンはAccountパラメータに含める（ex. ab12345.ap-southeast-1）
aws ssm put-parameter --name "/Snowflake/Dev/Account" --value "Account.Region" --type String
aws ssm put-parameter --name "/Snowflake/Dev/User" --value "User" --type String
aws ssm put-parameter --name "/Snowflake/Dev/Database" --value "Database" --type String
aws ssm put-parameter --name "/Snowflake/Dev/Warehouse" --value "Warehouse" --type String
aws ssm put-parameter --name "/Snowflake/Dev/Password" --value "Password" --type SecureString --key-id alias/aws/ssm

ちなみに、Parameter Storeからのパラメーターの取得にはianlopshire/go-ssm-configを使ってみました。

JWT認証で接続

gosnowflakeはJWT認証にも対応しています（ただし、現時点でPreviewのステータスで、プロダクション環境での利用は非推奨になっています）。

キーペアの作成

JWTの署名に使用するキーペアを生成します。キーペアはPKCS#8形式で作成します。

  • プライベートキー生成
$ openssl genpkey -algorithm RSA \
-pkeyopt rsa_keygen_bits:2048 \
-pkeyopt rsa_keygen_pubexp:65537 | \
openssl pkcs8 -topk8 -nocrypt -outform der > rsa-2048-private-key.p8
  • パブリックキー生成
$ openssl pkey -pubout -inform der -outform der \
-in rsa-2048-private-key.p8 \
-out rsa-2048-public-key.spki

AWS Secrets Managerにプライベートキーを保存

プライベートキーはバイナリファイルです。このため、今回はキーファイルの保存先としてParameter Storeではなくバイナリファイルが保存可能なAWS Secrets Managerを利用します。

AWS CLIでプライベートキーをAWS Secrets Managerに保存します。

$ aws secretsmanager create-secret --name snowflake-jwt-privatekey \
      --description "private key for SnowFlake JWT authentication" \
      --kms-key-id "XXXXXXXX-XXXX-XXXX-XXXXXXXXXXXXXXXXX" \
      --secret-binary fileb://rsa-2048-private-key.p8

Snowflakeにパブリックキーを登録

Snowflakeのユーザーに対してパブリックキーを登録します。

パブリックキーをbase64でエンコードします。

base64 ./rsa-2048-private-key.p8

Snowflakeの管理コンソールを開き、Worksheetsから以下のクエリを実行します。RSA_PUBLIC_KEYにはbase64でエンコードしたパブリックキーを指定します。

alter user <your_user_name> set rsa_public_key='XXXXXXXXXX';

snowflake-exec-query-alter-user-set rsa_public_key

サンプルプログラム

接続パラメーター（Config）で、パスワードの代わりにAuthenticator: sf.AuthTypeJwtとPrivateKey: <private key>,を指定します。

package main

import (
    "crypto/rsa"
    "crypto/x509"
    "database/sql"
    "errors"
    "flag"
    "fmt"
    "log"
    "os"

    "github.com/olekukonko/tablewriter"

    "github.com/aws/aws-sdk-go/aws"

    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/secretsmanager"
    "github.com/aws/aws-sdk-go/service/ssm"
    ssmconfig "github.com/ianlopshire/go-ssm-config"

    sf "github.com/snowflakedb/gosnowflake"
)

type Config struct {
    Account   string `ssm:"Account" required:"true"`
    User      string `ssm:"User" required:"true"`
    Password  string `ssm:"Password"`
    Database  string `ssm:"Database"`
    Warehouse string `ssm:"Warehouse"`
}

func getDSN() (string, *sf.Config, error) {

    sess := session.Must(session.NewSession())

    provider := &ssmconfig.Provider{
        SSM: ssm.New(sess, aws.NewConfig().WithRegion("ap-northeast-1")),
    }

    smSvc := secretsmanager.New(sess, aws.NewConfig().WithRegion("ap-northeast-1"))
    input := &secretsmanager.GetSecretValueInput{
        SecretId: aws.String("snowflake-jwt-privatekey"),
    }

    result, err := smSvc.GetSecretValue(input)
    if err != nil {
        return "", nil, err
    }

    keyInterface, err := x509.ParsePKCS8PrivateKey(result.SecretBinary)
    if err != nil {
        return "", nil, err
    }

    key, ok := keyInterface.(*rsa.PrivateKey)
    if !ok {
        return "", nil, errors.New("not RSA private key")
    }

    var c Config
    err = provider.Process("/Snowflake/Dev/", &c)
    if err != nil {
        return "", nil, err
    }

    cfg := &sf.Config{
        Account:       c.Account,
        User:          c.User,
        Authenticator: sf.AuthTypeJwt,
        PrivateKey:    key,
        Database:      c.Database,
        Warehouse:     c.Warehouse,
    }

    dsn, err := sf.DSN(cfg)
    return dsn, cfg, err
}

func run(dsn string) error {
    db, err := sql.Open("snowflake", dsn)
    if err != nil {
        log.Fatalf("failed to connect. %v, err: %v", dsn, err)
        return err
    }
    defer db.Close()

    query := "select email from emp_basic where email like '%.uk';"
    rows, err := db.Query(query)
    if err != nil {
        log.Fatalf("failed to run a query. %v, err: %v", query, err)
        return err
    }
    defer rows.Close()

    table := tablewriter.NewWriter(os.Stdout)
    table.SetHeader([]string{"Email"})
    var email string

    for rows.Next() {
        err := rows.Scan(&email)
        if err != nil {
            log.Fatalf("failed to get result. err: %v", err)
        }
        table.Append([]string{email})
    }
    if rows.Err() != nil {
        fmt.Printf("ERROR: %v\n", rows.Err())
        return rows.Err()
    }
    table.Render()
    return nil
}

func main() {
    if !flag.Parsed() {
        flag.Parse()
    }

    dsn, cfg, err := getDSN()
    if err != nil {
        log.Fatalf("failed to create DSN from Config: %v, err: %v", cfg, err)
        os.Exit(1)
    }

    err = run(dsn)
    if err != nil {
        log.Fatalf("failed to execute a query, err: %v", err)
        os.Exit(1)
    }
}

さいごに

以上、Go Snowflake Driver(gosnowflake)について簡単に紹介しました。現時点でgosnowflakeはPUTとGETコマンドに対応していないため、データのロード・アンロードには利用できません。ETLスクリプト開発には Snowflake Connector for Python等を使った方がよさそうですが、Snowflakeをバックエンドにした簡単なツール類の開発であればgosnowflakeでも十分まかなえそうです。他のコネクターやドライバーについても試してみたいと思います。

明日16日目は石川によるSnowflake 半構造化データサポートの解説」の予定です。お楽しみに。

Share this article

facebook logohatena logotwitter logo

関連記事

【CDI】Informatica Cloud Data Integrationで、ソースのスキーマを自動的に読み込めるという「スキーマ変更処理」を検証してみた

【CDI】Informatica Cloud Data Integrationで、ソースのスキーマを自動的に読み込めるという「スキーマ変更処理」を検証してみた

User avatar
渡部晃季
2023.05.02
Redshiftで各テーブルのカラム毎の利用ブロック数を確認する

Redshiftで各テーブルのカラム毎の利用ブロック数を確認する

User avatar
k.uehara
2023.03.07
[レポート]AWSでデータメッシュアーキテクチャを構築する#ANT336 #reinvent

[レポート]AWSでデータメッシュアーキテクチャを構築する#ANT336 #reinvent

User avatar
nayu.t.s
2023.01.01
dbt Coalesce 2022と来年の告知

dbt Coalesce 2022と来年の告知

User avatar
ちゃあこ
2022.10.28
Classmethod's white logo
Facebook's logo
X's logo
YouTube's logo

© Classmethod, Inc. All rights reserved.