ACID特性 (ACID Property)

SnowflakeはS3, Blob, GCSを外部ステージとして設定し、データをロードする機能を備えています。 Snowflakeは各クラウドストレージとの接続方法として「ストレージ統合」の使用を推奨しています。 今回、GCSとのストレージ統合を設定し、外部ステージのロード先としてみました。 その手順等を感想を混ぜつつ書いてみようと思います。 参考にしたSnowflake公式ドキュメントは以下です。 ストレージ統合の設定方法がステップbyステップで示されています。 Google Cloud Storageの統合の構成 この記事は、公式が説明するステップに従って書いていきます。 [arst_toc tag=\"h4\"] GCSとストレージ統合 一般に、各ストレージサービスと連携する際、接続に必要な接続情報が必要です。 接続情報を自力で保管し使用する場合、セキュリティリスクに晒されるため、 Snowflakeはよりセキュアな方法として「ストレージ統合」機能を提供しています。 現在、Snowflakeは「ストレージ統合」機能のみを推奨しています。 各クラウドプロバイダ毎に実装方法は異なり、GCSの場合、Service Accountを使用します。 クラウドプロバイダ間で概念が異なるため 「AWSでいうところのXXXだよね」 は誤解を生みますが、 簡単に言えば、AWSのIAM、Azureのサービスプリンシパル、が相当すると思います。(本当??) 以下の図はその概念図です。(公式直リンク) Snowflakeは、ストレージ統合オブジェクトをGCS用のサービスアカウントと紐付けます。 このサービスアカウントは裏でSnowflakeが作成します。 言い換えると、GCSの認証責任をSnowflakeが作成するサービスアカウントに委任します。 Snowflake上のロード・アンロード操作の裏で透過的にサービスアカウントの権限参照が行われます。 Snowflakeでストレージ統合を作成する Snowflake上でストレージ統合を作る方法は以下の通りです。 実行にはACCOUNTADMINロールか、CREATE INTEGRATION権限が必要です。 ストレージ統合を作成する段階では、GCP側の権限設定は不要です。 CREATE STORAGE INTEGRATION gcs_int TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = \'GCS\' ENABLED = TRUE STORAGE_ALLOWED_LOCATIONS = (\'gcs://mybucket1/path1/\', \'gcs://mybucket2/path2/\') STORAGE_BLOCKED_LOCATIONS = (\'gcs://mybucket1/path1/sensitivedata/\', \'gcs://mybucket2/path2/sensitivedata/\'); 自動生成されたサービスアカウントの情報を取得 CREATE INTEGRATIONによって作られたSnowflake用のサービスアカウント情報を取得します。 オブジェクトの詳細情報を取得する DESC コマンドにより取得できます。 DESC STORAGE INTEGRATION gcs_int; +-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+ | property | property_type | property_value | property_default | +-----------------------------+---------------+-----------------------------------------------------------------------------+------------------| | ENABLED | Boolean | true | false | | STORAGE_ALLOWED_LOCATIONS | List | gcs://mybucket1/path1/,gcs://mybucket2/path2/ | [] | | STORAGE_BLOCKED_LOCATIONS | List | gcs://mybucket1/path1/sensitivedata/,gcs://mybucket2/path2/sensitivedata/ | [] | | STORAGE_GCP_SERVICE_ACCOUNT | String | service-account-id@project1-123456.iam.gserviceaccount.com | | +-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+ STORAGE_GCP_SERVICE_ACCOUNTプロパティが、作成されたサービスアカウントです。 バケットオブジェクトにアクセスするためのサービスアカウント権限を付与する 最後に、サービスアカウントがバケットへアクセスするための権限設定を行います。 ここで初めて、GCPコンソール上での操作が必要となります。 カスタムIAMロールの作成 以下の権限を持つIAMロールを作成します。今回はロードのみを行うため以下を設定します。 パージを行う場合はobjects.delte、アンロードを行う場合はobjects.createが必要です。 storage.buckets.get storage.objects.delete storage.objects.get storage.objects.list サービスアカウントへIAMロールを割り当てる GCPのやり方に従って、サービスアカウントへIAMロールを割り当てます。 バケットに対して、サービスアカウントのアクセスを許可し、その許可内容をIAMロールで指定します。 Azureと比較するとAWSとGCPは近いなと感じます。 外部ステージの作成 まず、ステージオブジェクトを格納するSnowflakeデータベースへの権限付与を行います。 GRANT USAGE ON DATABASE mydb TO ROLE myrole; GRANT USAGE ON SCHEMA mydb.stages TO ROLE myrole; GRANT CREATE STAGE ON SCHEMA mydb.stages TO ROLE myrole; GRANT USAGE ON INTEGRATION gcs_int TO ROLE myrole; 次にストレージ統合を指定して外部ステージオブジェクトを作成します。 USE SCHEMA mydb.stages; CREATE STAGE my_gcs_stage URL = \'gcs://mybucket1\' STORAGE_INTEGRATION = gcs_int FILE_FORMAT = my_csv_format; 最後に疎通試験を行います。 copy into testtable from @my_gcs_stage pattern=\'testdata.csv\'; まとめ Snowflakeの公式ドキュメントの通りにGCSとのストレージ統合を作成しました。 また、作成したストレージ統合上に外部ステージを設定し、ロードが出来ることを確認しました。

Snowpipeは外部ステージ上に置かれたファイルを自動的にSnowflakeテーブルにロードする仕組み。 クラウドプロバイダの機能が透過的に利用されるため、その仕組みを意識する必要がない。 AWS,Azure,GCPがサポートされている。 Snowpipeを設定する際に、登場するオブジェクトにどう権限設定したら良いのかいつも悩む。 オレオレ設定をしてガバくなってないか、調べ直してみて書いてみる。 セキュリティを構成する SYSADMINでリソースを作って、SYSADMINにAssumeRoleして操作をする、というのは良くない。 別にロールを作ってそこに必要な権限を集めると良さそう。 既に外部ステージが存在し、そこに続々とファイルが配置されている。 また、既にSnowflakeにテーブルが存在し、外部ステージに到着するファイルを読み込みたい。 このシチュエーションで、外部ステージとSnowflakeテーブルを結ぶパイプを作成する。 パイプと関連する全リソースにアクセス可能なロールは以下のように作る。 パイプから見れば、外部ステージ、テーブルは既存のオブジェクトであり、 使用する、つまり USAGE が必要。またデータ取得のため、外部ステージから READ が必要。 同様に、パイプから見れば、データベース、スキーマは既存のオブジェクトであり、 USAGEが必要。またデータ投入のため、テーブルに対して INSERT が必要。 パイプがテーブル内のデータを参照して処理をするため テーブルに大して SELECTが必要。 例えば SYSADMIN が PIPE を CREATE した場合、その所有者は SYSADMIN になるが、 これを繰り返すと、SYSADMIN が全ての所有者になってしまう。 今回、Snowpipeに関わるオブジェクトを操作・閲覧可能なロールを作る、という話なので、 新ロールを PIPE の所有者にする。

[mathjax] マルコフ連鎖モンテカルロ法。2変量正規分布からGibbs Samplingする方法を考えてみました。 式を流してもよくわからないので、行間ゼロで理解できるまで細切れにして書いてみます。 Gibbs Sampling 無茶苦茶一般的に書かれている書籍だと以下みたいになっている。 ステップごとに(theta=(theta_1,theta_2,cdots,theta_n))の各要素を、 その要素以外の要素の条件付き確率でランダムに発生させる。 begin{eqnarray} theta^0 &=& (theta_1^0,theta_2^0,cdots,theta_n^0) \\ theta_i^{t+1} &sim& p(theta_i|theta_1^{t+1},cdots,theta_{i-1}^{t+1},theta_{i+1}^t,cdots,theta_n^t) end{eqnarray} 2次元空間であれば、ある点を決める際に、 片方の変数(theta_1)を固定して条件付き確率(P(theta_2|theta_1))を最大にするパラメタ(hat{theta_2})を決める。 その時点で((theta_1,hat{theta_2}))が決まる。 次は変数(hat{theta_2})を固定して条件付き確率(P(theta_1|hat{theta_2}))を最大にするパラメタ(hat{theta_1})を決める。 2次元正規分布であれば、片方の確率変数を固定したときの条件付き確率が1変数の正規分布になるから、 これがすごくやりやすい。 Gibbs Samplingは、このように条件付き確率を計算できる必要がある。 2変量正規分布の条件付き確率 2変量正規分布の片方の確率変数を固定すると1変数の正規分布が出てきます。 山の輪切りにして出てきた正規分布の母数を調べたいのですが、導出が無茶苦茶大変そうです。 出てきた正規分布の母数について式変形すると出てくるようです。こちらを参考にさせて頂きました。 この関係式を利用してひたすら式変形していきます。 begin{eqnarray} f(x|y) &=& frac{f(x,y)}{f(x)} end{eqnarray} 今、確率変数(y=y\')を固定し確率変数(x)の正規分布を考えます。 2変量正規分布の分散共分散行列が(sum)であるとします。 begin{eqnarray} sum &=& begin{pmatrix} sigma_{xx} & sigma_{xy} \\ sigma_{yx} & sigma_{yy} end{pmatrix} end{eqnarray} 出てくる正規分布の平均は以下となります。 確率変数が(x)なのに固定した(y)が出てくるのがポイント。 begin{eqnarray} mu\' = bar{x} + frac{sigma_{xy}}{sigma_{yy}} (y\'-bar{y}) end{eqnarray} また、出てくる正規分布の分散は以下となります。 begin{eqnarray} sigma\'^2 &=& sigma_{xx} - frac{sigma_{xy}}{sigma_{yy}} sigma_{yx} end{eqnarray} Python実装例 実際にPythonコードにしてみた図です。2変量正規分布の条件付き確率さえわかってしまえば あとはコードに落とすだけです。 import numpy as np import matplotlib.pyplot as plt import scipy.stats as stats n_dim = 2 def gibbs_sampling(mu, sigma, sample_size): samples = [] start = [0, 0] samples.append(start) search_dim = 0 for i in range(sample_size): search_dim = 0 if search_dim == n_dim-1 else search_dim + 1 prev_sample = samples[-1][:] s11 = sigma[search_dim][search_dim] s12 = sigma[search_dim][search_dim - 1] s21 = sigma[search_dim -1 ][search_dim] s22 = sigma[search_dim - 1][search_dim - 1] mu_x = mu[search_dim] mu_y = mu[search_dim-1] _y = prev_sample[search_dim - 1] new_mean = mu_x + s12/float(s22) * (_y - mu_y) new_sigma = s11 - s12/float(s22) * s21 sample_x = np.random.normal(loc=new_mean, scale=np.power(new_sigma, .5), size=1) prev_sample[search_dim] = sample_x[0] samples.append(prev_sample) return np.array(samples) nu = np.ones(2) covariance = np.array([[0.5, 0.5], [0.5, 3]]) samples = gibbs_sampling(nu, covariance, 1000) fig, ax1 = plt.subplots(figsize=(6, 6)) ax1.scatter(sample[:, 0], sample[:, 1], marker=\"o\", facecolor=\"none\", alpha=1., s=30., edgecolor=\"C0\", label=\"Samples\" ) 実行結果