簡介

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

創(chuàng)新互聯(lián)是一家專業(yè)提供文安企業(yè)網(wǎng)站建設(shè),專注與成都網(wǎng)站設(shè)計、做網(wǎng)站、H5場景定制、小程序制作等業(yè)務(wù)。10年已為文安眾多企業(yè)、政府機(jī)構(gòu)等服務(wù)。創(chuàng)新互聯(lián)專業(yè)網(wǎng)站制作公司優(yōu)惠進(jìn)行中。

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

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

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

架構(gòu)特點

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

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

Snowpark執(zhí)行

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

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

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

要檢索Salesforce_Account表的前10行，我們可以簡單地執(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會把代碼轉(zhuǎn)換成SQL語句并交給Snowflake執(zhí)行：

Scala

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

在我們的VSCode IDE中的輸出看起來像這樣:

我們也可以過濾某些行并執(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計算服務(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)。在這個例子中，我們有SALESFORCE_ORDER表，記錄了由Salesforce賬戶產(chǎn)生的賬單數(shù)據(jù)，我們可以將這些數(shù)據(jù)拉到DataFrame中，并將它們與賬戶記錄連接起來：

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語句，然后推送給Snowflake數(shù)據(jù)倉庫進(jìn)行計算。在VSCode中輸出如下：

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

Scala

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

生成的SQL語句看起來就像這樣：

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會創(chuàng)建一個新表或者替換掉已存在的舊表，來存儲生成的數(shù)據(jù)：

結(jié)語

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

譯者介紹

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

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

當(dāng)前標(biāo)題：使用基于Snowflake的SnowparkDataFrames進(jìn)行數(shù)據(jù)處理
轉(zhuǎn)載來源：http://uogjgqi.cn/article/dpdgegj.html