Databricks（Spark）のPysparkにてsnowflakeをソースとしてSparkデータフレームとSparkテーブルを作成する方法

## 概要
Databirkcs（Spark）からsnowflakeへ接続する方法として、jdbcによる方法とSparkコネクターによる方法があり、この記事では2つの方法にてSparkデータフレームとSparkテーブルを作成する方法を紹介します。

SnowflakeのドキュメントにてSparkコネクターを利用することが推奨されており、Sparkコネクターを優先して利用したほうがよさそうです。
>Snowflake JDBC ドライバーと組み合わせたコネクターは2つのシステム間で大量のデータを転送するために最適化されているため、Spark用のSnowflakeコネクターを使用することをお勧めします。また、SparkからSnowflakeへのクエリプッシュダウンをサポートすることにより、パフォーマンスが向上します。

引用元：[Sparkコネクターの概要 — Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/spark-connector-overview.html)

Sparkコネクターとjdbcでは、設定するパラメーターが微妙に異なるので注意してください。
Sparkコネクターで設定するパラメーターについては、[Sparkコネクタの使用 — Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/spark-connector-use.html#required-connection-options)にて説明されております。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7509a4b5-62a5-2581-ca8a-0b4084fb75f9.png)

引用元：[Sparkコネクタの使用 — Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/spark-connector-use.html#required-connection-options)

jdbcで設定するパラメーターについては、[JDBC ドライバーを構成する — Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/jdbc-configure.html#connection-parameters)にて説明されております。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/48a98d99-e32f-eb70-4011-0465c7d266c7.png)

引用元：[JDBC ドライバーを構成する — Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/jdbc-configure.html#connection-parameters)

コードを含むノートブックの実行結果をGithub pagesにて公開しております。

https://manabian-.github.io/databricks_tecks_for_qiita/#create_df_table_from_snowflake/create_df_table_from_snowflake.html

実際に試したい方は、下記のファイルをインポートしてください。

```URL
https://github.com/manabian-/databricks_tecks_for_qiita/blob/main/tecks/create_df_table_from_snowflake/create_df_table_from_snowflake.dbc
```

## 実施手順
### SparkコネクターによりSparkデータフレームを作成する方法
```python
# ドライバーを指定。Databrcksでは、"snowflake"と指定することでも可能
driver = 'net.snowflake.spark.snowflake'
# driver = 'snowflake'
 
# 接続オプションを、sfOptionsに格納。オプションについては、下記ドキュメントに記載あり。
# https://docs.snowflake.com/ja/user-guide/spark-connector-use.html#required-connection-options
sfOptions = {
 "sfURL" : "AAAA.southeast-asia.azure.snowflakecomputing.com",
 "sfUser" : "user",
 "sfPassword" : "password",
 "sfRole" : "SYSADMIN",
 "sfDatabase" : "SNOWFLAKE_SAMPLE_DATA",
 "sfSchema" : "TPCH_SF1",
 "sfWarehouse" : "COMPUTE_WH",
 "column_mapping" : "name",
 "sfTimezone" : "Asia/Tokyo",
}
 
# データの取得方法と取得元を指定
### dbtable（テーブルを指定する方法）、もしくは、query（SQLを指定する方法）を指定可能
how_to_read = 'dbtable'
table_or_query = 'NATION'
# how_to_read = 'query'
# table_or_query = 'select * from NATION'
    
## データフレームを作成
df = (spark.read
          .format(driver)
          .options(**sfOptions)
          .option(how_to_read, table_or_query)
          .load()
     )
 
## データフレームを表示
df.limit(10).display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a0c87088-f5be-9dc0-2a40-9cd843aa1478.png)


### jdbcによりSparkデータフレームを作成する方法
```python
# ドライバーを指定。
driver = 'net.snowflake.client.jdbc.SnowflakeDriver'
 
# 接続オプションを指定。指定可能なオプションについては、下記ドキュメントに記載あり。
# https://docs.snowflake.com/ja/user-guide/jdbc-configure.html#connection-parameters
 
url  = "jdbc:snowflake://AAAA.southeast-asia.azure.snowflakecomputing.com"
 
user = 'username'
password = 'password'
role = 'SYSADMIN'
 
warehouse = 'COMPUTE_WH'
 
db = 'SNOWFLAKE_SAMPLE_DATA'
schema = 'TPCH_SF1'
 
 
# データの取得方法と取得元を指定
### dbtable（テーブルを指定する方法）、もしくは、query（SQLを指定する方法）を指定可能
how_to_read = 'dbtable'
table_or_query = "NATION"
# how_to_read = "query"
# table_or_query = "select * from NATION"
 
## データフレームを作成
df = (spark.read
          .format("jdbc")
          .option("driver", driver)
          .option("url", url)
          .option("user", user)
          .option("password", password)                  
          .option("role", role)
          .option("warehouse", warehouse)
          .option("db", db)
          .option("schema", schema)
          .option(how_to_read, table_or_query)          
          .load()
     )
 
## データフレームを表示
df.limit(100).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/18302664-87c9-fcf2-7906-cf427b8aa7ce.png)

### SparkコネクターによりSparkテーブルを作成する方法
```python
# Sparkコネクター経由のSparkテーブルを作成する方法
## ドライバーを指定。Databrcksでは、"snowflake"と指定することでも可能
driver = 'net.snowflake.spark.snowflake'
# driver = 'snowflake'
 
# 接続オプションを、sfOptionsに格納。オプションについては、下記ドキュメントに記載あり。
# https://docs.snowflake.com/ja/user-guide/spark-connector-use.html#required-connection-options
# オプションの値を、"（ダブルクオーテーション）で囲む必要がある。
url  = '"AAAA.southeast-asia.azure.snowflakecomputing.com"'
 
warehouse = '"COMPUTE_WH"'
 
user = '"user"'
password = '"password"'
role = '"SYSADMIN"'
 
db = '"SNOWFLAKE_SAMPLE_DATA"'
schema = '"TPCH_SF1"'
 
column_mapping = '"name"'
 
# データの取得方法と取得元を指定
### dbtable（テーブルを指定する方法）を指定可能
how_to_read = 'dbtable'
table = '"NATION"'
 
spark_table_name = spark_database_name + '.NATION_from_snowflake_with_sparkconnector'
 
# テーブルをドロップ
spark.sql(f'drop table if exists {spark_table_name}')
 
# DDL文を作成
ddl__create_table = f'''
create table {spark_table_name}
using {driver}
options ( 
  sfURL {url}, 
  sfWarehouse {warehouse},
  sfUser {user}, 
  sfPassword {password},
  sfRole {role},
  sfDatabase {db},
  sfSchema {schema},
  {how_to_read} {table},
  column_mapping {column_mapping}
  )
'''
 
# DDL文の実行
spark.sql(ddl__create_table)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ca020e57-464f-a7c4-6fee-cb7038262283.png)


### jdbcによりSparkテーブルを作成する方法
```python
# jdbc経由のSparkテーブルを作成する方法
## ドライバーを指定。Databrcksでは、"snowflake"と指定することでも可能
driver = 'org.apache.spark.sql.jdbc'
# driver = 'snowflake'
 
## 接続オプションを、sfOptionsに格納。オプションについては、下記ドキュメントに記載あり。
## https://docs.snowflake.com/ja/user-guide/spark-connector-use.html#required-connection-options
## オプションの値を、"（ダブルクオーテーション）で囲む必要がある。
url  = '"jdbc:snowflake://AAAA.southeast-asia.azure.snowflakecomputing.com"'
 
warehouse = '"COMPUTE_WH"'
 
user = '"username"'
password = '"password"'
role = '"SYSADMIN"'
 
db = '"SNOWFLAKE_SAMPLE_DATA"'
schema = '"TPCH_SF1"'
dbtable = '"lineitem"'
 
# データの取得方法と取得元を指定
### dbtable（テーブルを指定する方法）を指定可能
how_to_read = 'dbtable'
table = '"NATION"'
 
# Sparkのテーブル名を指定
spark_table_name = spark_database_name + '.NATION_from_snowflake_with_jdbc'
 
# DDL文を作成
ddl__create_table = f'''
create table if not exists {spark_table_name}
using {driver}
options ( 
  url {url}, 
  warehouse {warehouse},
  user {user}, 
  password {password},
  role {role},
  db {db},
  schema {schema},
  {how_to_read} {table}
  )
'''
 
# DDL文を実行
spark.sql(ddl__create_table)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fb5ed3c1-03c8-bbe0-116a-84ca1456f479.png)