Snowflake からGoogle Map API をたたきに行く方法

概要

Snowflake からGoogle Map API をたたき，住所情報から緯度経度情報への変換や移動時間等を取得する方法について解説する．Google Map API 側での事前準備はこちらの記事を参考にするとよい．Google Maps APIを使用するには，APIキーを事前に取得しておく必要がある．毎月200ドル分は無料で使用することが可能だが，それを超えると利用料金を支払う必要があるため，利用料をモニタリングしながら実施することをお勧めする．

Snowflake 側での対応

Google Map API に接続するための設定

Network Ruleの作成

Snowflake から外部に接続するためのネットワークルールを設定する．今回は，"maps.googleapis.com"へのアクセスのルールを定義する．SQLのワークシートで，以下SQLを実行する．

CREATE OR REPLACE NETWORK RULE <DATABASE_NAME>.<SCHEMA_NAME>.google_map_api_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('maps.googleapis.com');

Secretの作成

事前にGoogle Map API 側で取得した，APIキーの認証情報を呼び出すSecret を設定する．

CREATE OR REPLACE SECRET <DATABASE_NAME>.<SCHEMA_NAME>.google_map_api_token
  TYPE = GENERIC_STRING
  SECRET_STRING = '<GOOGLEMAPAPIKEY>';

External Access Integrationの作成

先ほど作成したネットワークルールとSecretの情報を用いて，外部に接続するための外部ネットワークロケーションを設定する．この機能は2023年にリリースされており，この機能により，他クラウドとの連携が柔軟にできるようになった．

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION google_map_api_access
  ALLOWED_NETWORK_RULES = (<DATABASE_NAME>.<SCHEMA_NAME>.google_map_api_rule)
  ALLOWED_AUTHENTICATION_SECRETS = (<DATABASE_NAME>.<SCHEMA_NAME>.google_map_api_token)
  ENABLED = true;

Google Map API をたたくためのプログラムの定義

住所を緯度・経度に変換する関数

住所情報を入力すると緯度・経度情報に変換する関数である．先ほど作成したExternal Access Integration の設定等を使っている．

CREATE OR REPLACE FUNCTION <DATABASE_NAME>.<SCHEMA_NAME>.geocode_address(address STRING)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
HANDLER = 'main'
EXTERNAL_ACCESS_INTEGRATIONS = (google_map_api_access)
PACKAGES = ('requests')
SECRETS = ('cred' = <DATABASE_NAME>.<SCHEMA_NAME>.google_map_api_token)
AS
$$
import _snowflake
import requests
import json

def main(address):
    # APIキーの取得
    api_key = _snowflake.get_generic_secret_string('cred')
    
    # Google Maps APIのURL
    url = f"https://maps.googleapis.com/maps/api/geocode/json?address={address}&key={api_key}"
    
    # リクエストを送信
    response = requests.get(url)
    data = response.json()
    
    # 正常なレスポンスを受け取った場合
    if data["status"] == "OK":
        lat = data["results"][0]["geometry"]["location"]["lat"]
        lng = data["results"][0]["geometry"]["location"]["lng"]
        return {"lat": lat, "lng": lng}
    else:
        return None
$$;

以下，実行結果の一例である．
SQLを実行すると，"東京都千代田区丸の内1丁目"の住所が"[3.568187180000000e+01, 1.397658469000000e+02]"の緯度・経度に変換される．

SELECT <DATABASE_NAME>.<SCHEMA_NAME>.geocode_address('東京都千代田区丸の内1丁目');

住所間の移動時間を取得する関数

2カ所の住所情報を入力すると住所間の移動時間が出力される関数である．先ほど作成したExternal Access Integration の設定等を使っている．

CREATE OR REPLACE FUNCTION <DATABASE_NAME>.<SCHEMA_NAME>.get_travel_time_matrix(addresses ARRAY)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
HANDLER = 'main'
EXTERNAL_ACCESS_INTEGRATIONS = (google_map_api_access)
PACKAGES = ('requests')
SECRETS = ('cred' = <DATABASE_NAME>.<SCHEMA_NAME>.google_map_api_token)
AS
$$
import _snowflake
import requests
import json

def main(addresses):
    # APIキーの取得
    api_key = _snowflake.get_generic_secret_string('cred')
    
    # 住所リストから緯度経度を取得
    origins = "|".join(addresses)
    destinations = "|".join(addresses)
    
    # Google Maps APIのURL
    url = f"https://maps.googleapis.com/maps/api/distancematrix/json?origins={origins}&destinations={destinations}&key={api_key}"
    
    # リクエストを送信
    response = requests.get(url)
    data = response.json()
    
    # 正常なレスポンスを受け取った場合
    dist_matrix = []
    if data["status"] == "OK":
        for row in data["rows"]:
            dist_matrix.append([element["duration"]["value"] for element in row["elements"]])
    else:
        return None
    
    return dist_matrix
$$;

以下，実行結果の一例である．
SQLを実行すると，"東京都千代田区丸の内1丁目"と"大阪府大阪市北区梅田"の2カ所の住所間の移動時間"[[0,21998],[21907,0]]"が出力される．

SELECT <DATABASE_NAME>.<SCHEMA_NAME>.get_travel_time_matrix(ARRAY_CONSTRUCT('東京都千代田区丸の内1丁目', '大阪府大阪市北区梅田'));