Google
Cloud Platform
Amazon Web Services
Microsoft AzureSnowflakeからのデータの読み取りと書き込み
Databricks は、Snowflake からのデータの読み取りと書き込みをサポートするために、Databricks Runtime に Snowflake コネクタを提供します。
実験段階
この記事で説明する構成は 実験的です。 実験的な機能は現状のまま提供され、Databricks による顧客テクニカル サポートを通じてサポートされることはありません。 完全なクエリ フェデレーション サポートを取得するには、代わりにレイクハウス フェデレーションを使用する必要があります。これにより、 DatabricksユーザーはUnity Catalog構文とデータガバナンス ツールを活用できるようになります。
DatabricksにおけるSnowflakeテーブルへのクエリー
Snowflakeへの接続を構成してから、データをクエリーできます。 開始する前に、クラスターが実行されている Databricks Runtime のバージョンを確認します。 次のコードは、Python、SQL、および Scala の構文例を示しています。
# The following example applies to Databricks Runtime 11.3 LTS and above.
snowflake_table = (spark.read
.format("snowflake")
.option("host", "hostname")
.option("port", "port") # Optional - will use default port 443 if not specified.
.option("user", "username")
.option("password", "password")
.option("sfWarehouse", "warehouse_name")
.option("database", "database_name")
.option("schema", "schema_name") # Optional - will use default schema "public" if not specified.
.option("dbtable", "table_name")
.load()
)
# The following example applies to Databricks Runtime 10.4 and below.
snowflake_table = (spark.read
.format("snowflake")
.option("dbtable", table_name)
.option("sfUrl", database_host_url)
.option("sfUser", username)
.option("sfPassword", password)
.option("sfDatabase", database_name)
.option("sfSchema", schema_name)
.option("sfWarehouse", warehouse_name)
.load()
)
/* The following example applies to Databricks Runtime 11.3 LTS and above. */
DROP TABLE IF EXISTS snowflake_table;
CREATE TABLE snowflake_table
USING snowflake
OPTIONS (
host '<hostname>',
port '<port>', /* Optional - will use default port 443 if not specified. */
user '<username>',
password '<password>',
sfWarehouse '<warehouse_name>',
database '<database-name>',
schema '<schema-name>', /* Optional - will use default schema "public" if not specified. */
dbtable '<table-name>'
);
SELECT * FROM snowflake_table;
/* The following example applies to Databricks Runtime 10.4 LTS and below. */
DROP TABLE IF EXISTS snowflake_table;
CREATE TABLE snowflake_table
USING snowflake
OPTIONS (
dbtable '<table-name>',
sfUrl '<database-host-url>',
sfUser '<username>',
sfPassword '<password>',
sfDatabase '<database-name>',
sfSchema '<schema-name>',
sfWarehouse '<warehouse-name>'
);
SELECT * FROM snowflake_table;
# The following example applies to Databricks Runtime 11.3 LTS and above.
val snowflake_table = spark.read
.format("snowflake")
.option("host", "hostname")
.option("port", "port") /* Optional - will use default port 443 if not specified. */
.option("user", "username")
.option("password", "password")
.option("sfWarehouse", "warehouse_name")
.option("database", "database_name")
.option("schema", "schema_name") /* Optional - will use default schema "public" if not specified. */
.option("dbtable", "table_name")
.load()
# The following example applies to Databricks Runtime 10.4 and below.
val snowflake_table = spark.read
.format("snowflake")
.option("dbtable", table_name)
.option("sfUrl", database_host_url)
.option("sfUser", username)
.option("sfPassword", password)
.option("sfDatabase", database_name)
.option("sfSchema", schema_name)
.option("sfWarehouse", warehouse_name)
.load()
ノートブックの例: Spark用Snowflakeコネクタ
次のノートブックは、Snowflake に対してデータを書き込んだり、Snowflake からデータを読み取ったりする方法の簡単な例を示しています。 詳細については、 Spark 用 Snowflake コネクタ をご参照ください。
ヒント
ノートブックで示されている シークレットを使用して、ノートブックでSnowflakeのユーザー名とパスワードを公開しないようにします。
ノートブックの例: モデルのトレーニング結果をSnowflakeに保存する
次のノートブックでは、Spark用Snowflakeコネクタを使用するためのベストプラクティスについて説明します。 Snowflakeにデータを書き込み、基本的なデータ操作にSnowflakeを使用し、Databricksで機械学習モデルをトレーニングし、結果をSnowflakeに書き戻します。
よくある質問（FAQ）
Spark DataFrame列がSnowflakeで同じ順序で表示されないのはなぜですか?
Spark用Snowflakeコネクタは、書き込まれるテーブル内の列の順序を考慮しません。DataFrame 列とSnowflake列の間のマッピングを明示的に指定する必要があります。 このマッピングを指定するには、 columnmap パラメーターを使用します。
Snowflakeに書き込まれた INTEGER データが DECIMALとして読み戻されるのはなぜですか?
Snowflakeは、すべての INTEGER タイプを NUMBERとして表すため、Snowflakeにデータを書き込んだり、Snowflakeからデータを読み取ったりすると、データ型が変更される可能性があります。 たとえば、SnowflakeではINTEGERとDECIMALが意味的に同等であるため、Snowflakeへの書き込み時にINTEGERデータをDECIMALに変換できます( Snowflakeの数値データ型 を参照)。