從 Snowflake 讀取和寫入數據
Azure Databricks 在 Databricks Runtime 中提供 Snowflake 連接器,以支援從 Snowflake 讀取和寫入數據。
重要
本文所述的設定為 實驗性。 實驗性功能是以目前方式提供,且透過客戶支援不支援 Databricks。 若要取得完整的查詢同盟支援,您應該改用 Lakehouse 同盟,這可讓您的 Azure Databricks 使用者利用 Unity 目錄語法和數據控管工具。
查詢 Azure Databricks 中的 Snowflake 數據表
您可以設定與 Snowflake 的連線,然後查詢數據。 開始之前,請先檢查叢集執行所在的 Databricks Runtime 版本。 下列程式代碼提供 Python、SQL 和 Scala 中的範例語法。
Python
# 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()
)
SQL
/* 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;
Scala
# 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 Python 筆記本
筆記本範例:將模型定型結果儲存至 Snowflake
下列筆記本會逐步解說使用 Snowflake Connector for Spark 的最佳做法。 它會將數據寫入 Snowflake、針對某些基本數據操作使用 Snowflake、在 Azure Databricks 中定型機器學習模型,並將結果寫回 Snowflake。
在 Snowflake 筆記本中儲存 ML 訓練結果
常見問題集 (FAQ)
為什麼我的Spark DataFrame資料行在 Snowflake 中的順序不相同?
適用於 Spark 的 Snowflake 連接器不會遵守寫入數據表中數據行的順序;您必須明確指定 DataFrame 與 Snowflake 資料行之間的對應。 若要指定此對應,請使用 columnmap 參數。
INTEGER
為什麼將資料寫入 Snowflake 讀取回 DECIMAL
?
Snowflake 會將所有 INTEGER
類型表示為 NUMBER
,當您將數據寫入 Snowflake 並從 Snowflake 讀取數據時,可能會導致數據類型變更。 例如,當寫入 Snowflake 時, INTEGER
可以將數據轉換成 DECIMAL
,因為在 INTEGER
DECIMAL
Snowflake 中和 在語意上相等(請參閱 Snowflake 數值數據類型)。
為什麼我的 Snowflake 數據表架構中的欄位一律大寫?
Snowflake 預設會使用大寫欄位,這表示數據表架構會轉換成大寫。