Snowflake上でPythonを実行して外部サービス（Instagram）からデータを取得する

下記の記事では、サーバーから直接Pythonを実行してInstagramのデータを取得する方法について紹介しました。

最近はクラウドサービスを使うことでアプリケーション側のみに注力してシステムを構築できるようになりました。以前は実行環境のサーバーを用意してPythonを実行していたところをSnowflake上から実行できるようになっています。こうすることで構成がシンプルかつサーバーの維持費を抑えられ、大変お手軽です。

一方で、クラウドサービスから外部へのアクセスについてはほぼ確実に制限がついているケースが多いです。有害なリソースに勝手にアクセスされてしまうと困りますからね。Snowflakeも例外ではありません。

本記事ではこの制約を解消するExternal Network Accessと呼ばれる機能を使って外部リソース（Instagram）のデータを取得する方法について紹介します！

本記事は元々、データパレード様のブログに私が執筆した技術ノウハウとして掲載したものです（下記）。Qiitaを通じてより多くの方に届けられれば、そして技術者の方と知見を共有し、フィードバックをいただく目的で許可をいただき掲載しました。

外部アクセスを許可設定をしよう！

冒頭述べたとおり、SnowflakeではExternal Network Accessと呼ばれる外部のアクセス機能があります。

公式ドキュメントでは詳細が書かれていますが、ここではより具体的な手順を紹介できればと思います。とはいえ、比較的おまじないっぽく機械的にSQLを記載＆実行する部分が多いのであまり身構えなくても大丈夫です！

外部アクセスのルールを作成する

まずは実行するSQLを掲載しておきます。

-- <1> 対象のデータベースおよびスキーマを指定
use DATA_PARADE.ICHIBANHANTEN;

-- <2> 許可したい接続先のドメイン
create or replace network rule instagram_rule
mode = egress
type = host_port
value_list = ('graph.facebook.com');

-- <3> シークレットの設定（今回は空文字でOK）
create or replace secret instagram_service_account 
type = GENERIC_STRING
SECRET_STRING = '';

-- <4> <2>と<3>でexternal access integrationを作成
create or replace external access integration instagram_access
allowed_network_rules = (instagram_rule)
allowed_authentication_secrets = (instagram_service_account)
enabled = true;

最終目的はexternal access integrationの設定（ルール）を作成することです。まずは、対象のデータベースおよびスキーマを指定します。今回はデータベース名「DATA_PARADE」、スキーマ名は「ICHIBANHANTEN」としています。

use DATA_PARADE.ICHIBANHANTEN;

次にアクセスしたい外部サービスのドメインを指定します。今回はInstagramのWeb APIを利用するので、「graph.facebook.com」を許可リストに追加します。

create or replace network rule instagram_rule
mode = egress
type = host_port
value_list = ('graph.facebook.com');

続いて、アクセス時に使うシークレットを設定しますが今回は不要なので空文字とします。

create or replace secret instagram_service_account 
type = GENERIC_STRING
SECRET_STRING = '';

最後に作成したルールを各々指定してexternal access integrationを作ります。

create or replace external access integration instagram_access
allowed_network_rules = (instagram_rule)
allowed_authentication_secrets = (instagram_service_account)
enabled = true;

Notebookで記述する

integrationの指定

では、前章で作成したintegration「instagram_access」を使ってPythonからInstagramのデータを取得してみましょう。

Notebookを開き、右上の３点リーダーから「Notebook Settings」を開きます。

上部のタブを「External access」に切り替え、先ほど作成したintegrationをONにします。

Pythonを記述する

続いて、Pythonのセルを開き次のように記述します。今回はアカウントのプロファイル情報を取得するAPIを使ってデータを取得します。

IG_USER_IDとACCESS_TOKENには適宜設定情報を入力してください。

import _snowflake
import json
import requests

API_VERSION = "v17.0"
IG_USER_ID = 12345678901234567
ACCESS_TOKEN = "ABCDE...."

class getInstagramData:

    def process(self):
        url = f"https://graph.facebook.com/{API_VERSION}/{IG_USER_ID}"
        endpoint = url + "/"
        params = {
            "access_token" : ACCESS_TOKEN,
            "fields" : "business_discovery.username(ichibanhanten.official){name,username,biography,follows_count,followers_count,media_count}"
        }
        responce = requests.get(endpoint, params = params).json()
        row = responce.get('business_discovery', [])
        return row
gid = getInstagramData()
dict = gid.process() # process実行結果をdict変数に格納

最後の行でdict変数に結果を格納しています。dict変数はJSON形式でデータが入っています。

SQLを記述する

では、前節で記述したPythonの実行結果をSQLでテーブルに格納していきます。

SnowflakeのNotebookの超便利なポイントとして、別セルに記述した変数やクエリ実行結果を他のセルから参照できます！この機能を使って、次のSQLセルでテーブルにデータを入れていきましょう。

まずは単純にSELECT文で内容を確認してみます。記述するSQLは下記のようになります。

select
    {{dict}}:"name"::string as name,
    {{dict}}:"username"::string as username,
    {{dict}}:"biography"::string as biography,
    {{dict}}:"follows_count"::integer as follows_count,
    {{dict}}:"followers_count"::integer as followers_count,
    {{dict}}:"id"::string as id,
    CURRENT_DATE::date

ポイントは３つです。

• Pythonセルに格納した変数（今回はdict）の値を２重中括弧{{}}で囲むことで参照可能
  例：{{dict}}

• JSON形式のデータはダブルコロン:で値を参照可能
  例：{{dict}}:"name"

• 型を指定できる。整数値の場合はinteger、日付の場合はdateなど
  例：{{dict}}:"name":string

ではinsert into select構文で結果をテーブルに格納していきます。

insert into TBL_INSTAGRAM_PROFILE select
    {{dict}}:"name"::string as name,
    {{dict}}:"username"::string as username,
    {{dict}}:"biography"::string as biography,
    {{dict}}:"follows_count"::integer as follows_count,
    {{dict}}:"followers_count"::integer as followers_count,
    {{dict}}:"id"::string as id,
    CURRENT_DATE::date

タスクとして登録する

では最後に、Pythonからデータを取得しテーブルへ格納するまでの流れをタスクとして登録し、１日一回や６時間に一回自動で実行するようにします。

SnowflakeのNotebookで記述した一連の流れをGUI操作で簡単にタスクを作成できます。右上のカレンダーアイコンから「Create schedule」をクリックします。

タスクの名前、実行タイミング（毎日0時0分など）およびロケールを指定しCreateボタンを押下します。

以上でタスク化は完了です！簡単ですね！

【コラム】WorkSheetで作ってみると？

本記事ではNotebookを使いました。一方で、WorkSheetを使ってSQLベースで記述することもできます。結論から言うと、構築する過程で若干使いづらい部分が多いです。

以下はその一例です。

SQLベースで記述する都合上、CREATE FUNCTION構文中にPythonの定義を行うことになります。

これだと、Python実行の出力が確認できず別のPython実行環境でデバッグを行わないといけません。

また、あくまで関数なので入力および出力引数ありきの設計になります。table(value variant)のように出力がテーブル型であることを明示的に示す必要があります。前章で紹介したように、JSON形式の出力を直感的にSQL側で操作できることは大きなメリットです。

Pythonのシンタックスハイライトがなくなるのも地味に辛い部分ですね…

さいごに

本記事では、Snowflake上でPythonを実行して外部サービス（Instagram）からデータを取得し、テーブルへ格納するまでの一連の設定方法を紹介しました。また、それらのプロセスをタスク化し、日次や月次単位で自動実行する方法を説明しました。

Notebookの機能ではPythonとSQLをそれぞれ記述できるため非常にシームレスに構築ができます。今回紹介できていないものもたくさんあるので、興味があれば調べてみても良いかもしれません。