簡(jiǎn)介

Snowpark是Snowflake一個(gè)新的開發(fā)庫(kù)，它提供了一個(gè)API讓用戶可以使用編程語(yǔ)言像Scala(后續(xù)也會(huì)有Java和Python)來(lái)代替SQL進(jìn)行數(shù)據(jù)處理。

Snowpark的核心概念是DataFrame(數(shù)據(jù)框)，它表示一組數(shù)據(jù)，就比如說(shuō)一些數(shù)據(jù)庫(kù)表的行，我們可以用最喜歡的工具通過(guò)面向?qū)ο蠡蛘吆瘮?shù)式編程的方式處理。Snowpark DataFrames的概念類似于Apache Spark或者Python中Pandas包的DataFrames的含義，是一種表格型的數(shù)據(jù)結(jié)構(gòu)。

開發(fā)者也可以創(chuàng)建自定義函數(shù)推送到Snowflake服務(wù)器，來(lái)更方便地處理數(shù)據(jù)。Snowpark的代碼執(zhí)行采用了惰性計(jì)算的方式，這減少了從Snowpark倉(cāng)庫(kù)到客戶端之間的數(shù)據(jù)流轉(zhuǎn)。

當(dāng)前版本的Snowpark可以運(yùn)行在Scala 2.12和JDK 8、9、10或11上。它現(xiàn)在處于公開預(yù)覽階段，可用于所有賬戶。

架構(gòu)特點(diǎn)

從架構(gòu)的角度來(lái)看，Snowpark客戶端類似于Apache Spark Driver程序。它執(zhí)行用戶在客戶端編寫的代碼并轉(zhuǎn)為SQL語(yǔ)句推送給Snowpark數(shù)據(jù)倉(cāng)庫(kù)，等Snowpark計(jì)算服務(wù)端處理完數(shù)據(jù)后，接收以DataFrame格式組成的返回結(jié)果。

廣義的說(shuō)，Snowpark數(shù)據(jù)倉(cāng)庫(kù)的操作可以分為兩類：轉(zhuǎn)換和執(zhí)行。由于轉(zhuǎn)換是延遲執(zhí)行的，因此它們不會(huì)觸發(fā)DataFrames數(shù)據(jù)的計(jì)算處理過(guò)程。像select(查詢)，filter(過(guò)濾)，sort(排序)，groupBy(分組)等等都屬于轉(zhuǎn)換范疇的操作。而執(zhí)行是正好相反的，它們會(huì)觸發(fā)對(duì)DataFrames數(shù)據(jù)的計(jì)算。Snowpark將針對(duì)DataFrame數(shù)據(jù)的SQL語(yǔ)句發(fā)送到服務(wù)端進(jìn)行計(jì)算，然后將結(jié)果返回給客戶端內(nèi)存。show，collect，take等都屬于執(zhí)行操作。

Snowpark執(zhí)行

在我們可以執(zhí)行任何Snowpark轉(zhuǎn)換和執(zhí)行之前，我們需要先連接到Snowpark數(shù)據(jù)倉(cāng)庫(kù)并建立會(huì)話。

Scala
object Main {
  def main(args: Array[String]): Unit = {
    // Replace the <placeholders> below.
    val configs = Map (
      "URL" -> "https://<SNOWFLAKE-INSTANCE>.snowflakecomputing.com:443",
      "USER" -> "<USERNAME>",
      "PASSWORD" -> "<PASSWORD>",
      "ROLE" -> "SYSADMIN",
      "WAREHOUSE" -> "SALESFORCE_ACCOUNT",
      "DB" -> "SALESFORCE_DB",
      "SCHEMA" -> "SALESFORCE"
    )
    val session = Session.builder.configs(configs).create
    session.sql("show tables").show()
  }
}

從Snowpark管理頁(yè)面上看，我們有一個(gè)SALESFORCE_DB數(shù)據(jù)庫(kù)和一個(gè)有3個(gè)表的SALESFORCE：SALESFORCE_ACCOUNT表表示來(lái)自Salesforce實(shí)例的賬戶，SALESFORCE_ORDER表存儲(chǔ)由這些賬戶發(fā)起的訂單，SALESFORCE_ACCOUNT_ORDER是一個(gè)關(guān)聯(lián)表，存儲(chǔ)關(guān)聯(lián)的查詢結(jié)果(我們?cè)谶@篇文章的后面會(huì)再論述這點(diǎn))。

要檢索Salesforce_Account表的前10行，我們可以簡(jiǎn)單地執(zhí)行以下DataFrame方法：

Scala

 // Create a DataFrame from the data in the "salesforce_account" table.
    val dfAccount = session.table("salesforce_account")
     // To print out the first 10 rows, call:
     dfAccount.show()

Snowpark會(huì)把代碼轉(zhuǎn)換成SQL語(yǔ)句并交給Snowflake執(zhí)行：

Scala

[main] INFO com.snowflake.snowpark.internal.ServerConnection - Execute query [queryID: XXXX]  SELECT  *  FROM ( SELECT  *  FROM (salesforce_account)) LIMIT 10

在我們的VSCode IDE中的輸出看起來(lái)像這樣:

我們也可以過(guò)濾某些行并執(zhí)行DataFrame的轉(zhuǎn)換(例如，選擇指定的列)：

Scala

    val dfFilter = session.table("salesforce_account").filter(col("type") === "Customer - Direct")
     dfFilter.show()
     val dfSelect = session.table("salesforce_account").select(col("accountname"), col("phone"))
     dfSelect.show()

Snowpark將生成相應(yīng)的SQL查詢，并將它們交給Snowflake計(jì)算服務(wù)器執(zhí)行：

[main] INFO com.snowflake.snowpark.internal.ServerConnection - Execute query [queryID: XXXX]  SELECT  *  FROM ( SELECT  *  FROM ( SELECT  *  FROM (salesforce_account)) WHERE ("TYPE" = 'Customer - Direct')) LIMIT 10
  
 [main] INFO com.snowflake.snowpark.internal.ServerConnection - Execute query [queryID: XXXX]  SELECT  *  FROM ( SELECT "ACCOUNTNAME", "PHONE" FROM ( SELECT  *  FROM (salesforce_account))) LIMIT 10

下面是在VSCode中的輸出：

Snowpark DataFrame API也允許DataFrames數(shù)據(jù)間的拼接關(guān)聯(lián)。在這個(gè)例子中，我們有SALESFORCE_ORDER表，記錄了由Salesforce賬戶產(chǎn)生的賬單數(shù)據(jù)，我們可以將這些數(shù)據(jù)拉到DataFrame中，并將它們與賬戶記錄連接起來(lái)：

Scala

    val dfOrder = session.table("salesforce_order")
     dfOrder.show()
     val dfJoin = dfAccount.join(dfOrder, col("sfdcid") === col("accountid")).select(col("accountname"), col("phone"),col("productname"), col("amount"))
     dfJoin.show()

Snowflake把DataFrame方法轉(zhuǎn)換為SQL語(yǔ)句，然后推送給Snowflake數(shù)據(jù)倉(cāng)庫(kù)進(jìn)行計(jì)算。在VSCode中輸出如下：

如果我們想持久化保存計(jì)算結(jié)果，可以使用saveAsTable這個(gè)方法：

Scala

 dfJoin.write.mode(SaveMode.Overwrite).saveAsTable("salesforce_account_order")

生成的SQL語(yǔ)句看起來(lái)就像這樣：

Scala

[main] INFO com.snowflake.snowpark.internal.ServerConnection - Execute query [queryID: XXXX]  CREATE  OR  REPLACE  TABLE salesforce_account_order AS  SELECT  *  FROM ( SELECT "ACCOUNTNAME", "PHONE", "PRODUCTNAME", "AMOUNT" FROM ( SELECT  *  FROM (( SELECT "ACCOUNTNAME" AS "ACCOUNTNAME", "PHONE" AS "PHONE", "TYPE" AS "TYPE", "SFDCID" AS "SFDCID" FROM ( SELECT  *  FROM (salesforce_account))) AS SNOWPARK_TEMP_TABLE_UKKLR6UCHN6POXL INNER JOIN ( SELECT "ACCOUNTID" AS "ACCOUNTID", "PRODUCTNAME" AS "PRODUCTNAME", "AMOUNT" AS "AMOUNT" FROM ( SELECT  *  FROM (salesforce_order))) AS SNOWPARK_TEMP_TABLE_36DEOZXTQJUYKLD ON ("SFDCID" = "ACCOUNTID"))))

隨后，Snowpark會(huì)創(chuàng)建一個(gè)新表或者替換掉已存在的舊表，來(lái)存儲(chǔ)生成的數(shù)據(jù)：

結(jié)語(yǔ)

Snowpark為數(shù)據(jù)處理提供了豐富的操作和工具。它允許用戶創(chuàng)建非常復(fù)雜的高級(jí)數(shù)據(jù)處理管道操作。將用戶自定義的代碼推到Snowflake數(shù)據(jù)倉(cāng)庫(kù)服務(wù)端，并通過(guò)減少不必要的數(shù)據(jù)傳輸，在數(shù)據(jù)端執(zhí)行，這是Snowpark的一個(gè)非常強(qiáng)大的特性。

譯者介紹

盧鑫旺，51CTO社區(qū)編輯，半路出家的九零后程序員。做過(guò)前端頁(yè)面，寫過(guò)業(yè)務(wù)接口，搞過(guò)爬蟲，研究過(guò)JS，有幸接觸Golang，參與微服務(wù)架構(gòu)轉(zhuǎn)型。目前主寫Java，負(fù)責(zé)公司可定制化低代碼平臺(tái)的數(shù)據(jù)引擎層設(shè)計(jì)開發(fā)工作。

原文標(biāo)題：Snowflake Data Processing With Snowpark DataFrames，作者：Istvan Szegedi