Snowpark UDFs In Scala

1.介紹

當你使用 Snowpark API 創(chuàng)建一個 UDF 時，Snowpark 庫會將你的 UDF 代碼序列化并上傳到一個Internal Stage。當你調(diào)用 UDF 時，Snowpark 庫會在數(shù)據(jù)所在的服務(wù)器上執(zhí)行你的函數(shù)。因此，數(shù)據(jù)不需要傳輸?shù)娇蛻舳耍瘮?shù)就能處理數(shù)據(jù)。

在你的自定義代碼中，你還可以調(diào)用打包在 JAR 文件中的代碼（例如，用于第三方庫的 Java 類）。

有以下另種方法創(chuàng)建UDF：

  • 你可以創(chuàng)建一個匿名的 UDF，并將該函數(shù)賦值給一個變量。只要該變量在作用域內(nèi)，你就可以使用該變量來調(diào)用該 UDF。
// Create and register an anonymous UDF (doubleUdf).
val doubleUdf = udf((x: Int) => x + x)
// Call the anonymous UDF.
val dfWithDoubleNum = df.withColumn("doubleNum", doubleUdf(col("num")))
  • 你可以創(chuàng)建一個帶有名稱的 UDF，并通過名稱來調(diào)用該 UDF。例如，如果你需要通過名稱調(diào)用一個 UDF，或者在后續(xù)的會話中使用該 UDF，你可以使用這種方式。
// Create and register a permanent named UDF ("doubleUdf").
session.udf.registerPermanent("doubleUdf", (x: Int) => x + x, "mystage")
// Call the named UDF.
val dfWithDoubleNum = df.withColumn("doubleNum", callUDF("doubleUdf", col("num")))

2.UDF中支持的數(shù)據(jù)類型

image.png

3.（注意）在具有App Trait的對象中創(chuàng)建UDF

Scala提供了一個App trait，你可以擴展它以將你的Scala對象轉(zhuǎn)化為可執(zhí)行程序。App trait提供了一個main方法，它會自動執(zhí)行對象定義體中的所有代碼。（對象定義體中的代碼實際上成為了主方法。）

擴展App trait的一個影響是，直到調(diào)用main方法之前，對象中的字段不會被初始化。如果你的對象擴展了App，并且你定義了一個使用之前初始化的對象字段的UDF，上傳到服務(wù)器的UDF定義將不包括對象字段的初始化值。

例如，假設(shè)你在對象中定義并初始化了一個名為myConst的字段，并在一個UDF中使用了該字段：

object Main extends App {
  ...
  // Initialize a field.
  val myConst = "Prefix "
  // Use the field in a UDF.
  // Because the App trait delays the initialization of the object fields,
  // myConst in the UDF definition resolves to null.
  val myUdf = udf((s : String) =>  myConst + s )
  ...
}

當Snowpark將UDF定義序列化并上傳到Snowflake時，myConst沒有被初始化，因此解析為null。因此，調(diào)用UDF會返回myConst的null值。

為了解決這個問題，你需要修改你的對象，不再擴展App trait，并為你的代碼實現(xiàn)一個獨立的main方法：

object Main {
  ...
  def main(args: Array[String]): Unit = {
    ... // Your code ...
  }
  ...
}

4.為UDF指定依賴

為了通過Snowpark API定義一個UDF，你必須調(diào)用Session.addDependency()來添加包含UDF所依賴的類和資源的文件（例如JAR文件、資源文件等）。Snowpark庫將這些文件上傳到一個Internal Stage，并在執(zhí)行UDF時將這些文件添加到類路徑中。
這樣做的目的是確保在執(zhí)行UDF時能夠訪問到所依賴的類和資源。

注意：如果你不希望在每次運行應(yīng)用程序時都上傳文件，可以將文件上傳到一個階段（stage）。在調(diào)用 addDependency 時，傳遞文件在階段中的路徑。這樣做的好處是，可以避免重復(fù)上傳文件，提高應(yīng)用程序的執(zhí)行效率。通過指定階段中文件的路徑，Snowpark庫將能夠在執(zhí)行UDF時訪問該文件。

  • add jar file in a stage
// Add a JAR file that you uploaded to a stage.
session.addDependency("@my_stage/<path>/my-library.jar")

5.創(chuàng)建匿名UDF

要創(chuàng)建一個匿名的UDF，你可以采取以下兩種方法之一：

  1. 調(diào)用 com.snowflake.snowpark.functions 對象中的 udf 函數(shù)，傳入匿名函數(shù)的定義。
  2. 調(diào)用 UDFRegistration 類中的 registerTemporary 方法，傳入匿名函數(shù)的定義。由于你注冊的是一個匿名UDF，所以必須使用不帶有名稱參數(shù)的方法簽名。

注意：當編寫多線程代碼（例如使用并行集合）時，建議使用 registerTemporary 方法注冊UDF，而不是使用 udf 函數(shù)。這可以避免出現(xiàn)找不到默認的Snowflake Session對象的錯誤。

// Create and register an anonymous UDF.
val doubleUdf = udf((x: Int) => x + x)
// Call the anonymous UDF, passing in the "num" column.
// The example uses withColumn to return a DataFrame containing
// the UDF result in a new column named "doubleNum".
val dfWithDoubleNum = df.withColumn("doubleNum", doubleUdf(col("num")))

以下示例創(chuàng)建了一個使用自定義類（LanguageDetector，用于檢測文本中使用的語言）的匿名UDF。該示例調(diào)用匿名UDF來檢測DataFrame中的text_data列中的語言，并創(chuàng)建一個新的DataFrame，其中包含一個額外的lang列，表示所使用的語言。

// Import the udf function from the functions object.
import com.snowflake.snowpark.functions._

// Import the package for your custom code.
// The custom code in this example detects the language of textual data.
import com.mycompany.LanguageDetector

// If the custom code is packaged in a JAR file, add that JAR file as
// a dependency.
session.addDependency("$HOME/language-detector.jar")

// Create a detector
val detector = new LanguageDetector()

// Create an anonymous UDF that takes a string of text and returns the language used in that string.
// Note that this captures the detector object created above.
// Assign the UDF to the langUdf variable, which will be used to call the UDF.
val langUdf = udf((s: String) =>
     Option(detector.detect(s)).getOrElse("UNKNOWN"))

// Create a new DataFrame that contains an additional "lang" column that contains the language
// detected by the UDF.
val dfEmailsWithLangCol =
    dfEmails.withColumn("lang", langUdf(col("text_data")))

6.創(chuàng)建具名UDF

如果你想通過名稱調(diào)用一個UDF（例如使用functions對象中的callUDF函數(shù)），或者如果你需要在后續(xù)會話中使用一個UDF，你可以創(chuàng)建和注冊一個命名的UDF。為此，可以使用UDFRegistration類中的以下方法之一：

  • registerTemporary：如果你只計劃在當前會話中使用該UDF。
  • registerPermanent：如果你計劃在后續(xù)會話中使用該UDF。

要訪問UDFRegistration類的對象，調(diào)用Session類的udf方法。
registerTemporary方法創(chuàng)建了一個臨時的UDF，你可以在當前會話中使用它。

// Create and register a temporary named UDF.
session.udf.registerTemporary("doubleUdf", (x: Int) => x + x)
// Call the named UDF, passing in the "num" column.
// The example uses withColumn to return a DataFrame containing
// the UDF result in a new column named "doubleNum".
val dfWithDoubleNum = df.withColumn("doubleNum", callUDF("doubleUdf", col("num")))

7.使用不可序列化的對象

當你為lambda或函數(shù)創(chuàng)建UDF時，Snowpark庫會對lambda閉包進行序列化并將其發(fā)送到服務(wù)器執(zhí)行。
如果被lambda閉包捕獲的對象不可序列化，Snowpark庫將拋出java.io.NotSerializableException異常。

如果出現(xiàn)這種情況，你可以采取以下兩種方法之一：

  1. 使對象可序列化。
  2. 將對象聲明為lazy val或使用@transient注解，以避免對對象進行序列化。
    例如：
// Declare the detector object as lazy.
lazy val detector = new LanguageDetector("en")
// The detector object is not serialized but is instead reconstructed on the server.
val langUdf = udf((s: String) =>
     Option(detector.detect(s)).getOrElse("UNKNOWN"))

8.編寫UDF的初始化代碼

如果你的UDF需要初始化代碼或上下文，你可以通過UDF閉包中捕獲的值來提供這些內(nèi)容。

下面的示例使用一個單獨的類來初始化三個UDF所需的上下文。

  • 第一個UDF在lambda內(nèi)部創(chuàng)建了該類的一個新實例，因此每次調(diào)用UDF時都會執(zhí)行初始化。
  • 第二個UDF捕獲了在客戶端程序中生成的該類的實例。在客戶端生成的上下文被序列化，并被UDF使用。請注意，為了使該方法有效，上下文類必須是可序列化的。
  • 第三個UDF捕獲了一個lazy val，因此上下文在第一次UDF調(diào)用時被惰性實例化，并在后續(xù)調(diào)用中被重用。即使上下文不可序列化，這種方法也有效。但是，并不能保證數(shù)據(jù)幀中的所有UDF調(diào)用都使用同一個惰性生成的上下文。
import com.snowflake.snowpark._
import com.snowflake.snowpark.functions._
import scala.util.Random

// Context needed for a UDF.
class Context {
  val randomInt = Random.nextInt
}

// Serializable context needed for the UDF.
class SerContext extends Serializable {
  val randomInt = Random.nextInt
}

object TestUdf {
  def main(args: Array[String]): Unit = {
    // Create the session.
    val session = Session.builder.configFile("/<path>/profile.properties").create
    import session.implicits._
    session.range(1, 10, 2).show()

    // Create a DataFrame with two columns ("c" and "d").
    val dummy = session.createDataFrame(Seq((1, 1), (2, 2), (3, 3))).toDF("c", "d")
    dummy.show()

    // Initialize the context once per invocation.
    val udfRepeatedInit = udf((i: Int) => (new Context).randomInt)
    dummy.select(udfRepeatedInit('c)).show()

    // Initialize the serializable context only once,
    // regardless of the number of times that the UDF is invoked.
    val sC = new SerContext
    val udfOnceInit = udf((i: Int) => sC.randomInt)
    dummy.select(udfOnceInit('c)).show()

    // Initialize the non-serializable context only once,
    // regardless of the number of times that the UDF is invoked.
    lazy val unserC = new Context
    val udfOnceInitU = udf((i: Int) => unserC.randomInt)
    dummy.select(udfOnceInitU('c)).show()
  }
}

9.在UDF中讀取文件

正如前面提到的，Snowpark庫會將UDF上傳并在服務(wù)器上執(zhí)行。如果你的UDF需要從文件中讀取數(shù)據(jù)，你必須確保文件與UDF一起上傳。

另外，如果文件的內(nèi)容在調(diào)用UDF之間保持不變，你可以編寫代碼，在第一次調(diào)用時加載文件，并在后續(xù)調(diào)用中不再加載。這可以提高UDF調(diào)用的性能。

example:

1.Add the file to a jar file
# Create a new JAR file containing data/hello.txt.
$ jar cvf <path>/myJar.jar data/hello.txt

2.Specify that the JAR file is a dependency
// Specify that myJar.jar contains files that your UDF depends on.
session.addDependency("<path>/myJar.jar")

3.In the UDF, call Class.getResourceAsStream to find the file in the classpath and read the file.
// Read data/hello.txt from myJar.jar.
val resourceName = "/data/hello.txt"
val inputStream = classOf[com.snowflake.snowpark.DataFrame].getResourceAsStream(resourceName)

注意：如果你不希望文件的內(nèi)容在UDF調(diào)用之間發(fā)生變化，可以將文件讀取操作放在一個lazy val中。這將導(dǎo)致文件加載代碼僅在第一次調(diào)用UDF時執(zhí)行，而不會在后續(xù)調(diào)用中執(zhí)行。

下面的示例定義了一個對象（UDFCode），其中包含一個函數(shù)作為UDF（readFileFunc）使用。該函數(shù)讀取文件data/hello.txt，該文件預(yù)計包含字符串"hello"。該函數(shù)將此字符串添加到作為參數(shù)傳入的字符串前面。

// Create a function object that reads a file.
object UDFCode extends Serializable {

  // The code in this block reads the file. To prevent this code from executing each time that the UDF is called,
  // the code is used in the definition of a lazy val. The code for a lazy val is executed only once when the variable is
  // first accessed.
  lazy val prefix = {
    import java.io._
    val resourceName = "/data/hello.txt"
    val inputStream = classOf[com.snowflake.snowpark.DataFrame]
      .getResourceAsStream(resourceName)
    if (inputStream == null) {
      throw new Exception("Can't find file " + resourceName)
    }
    scala.io.Source.fromInputStream(inputStream).mkString
  }

  val readFileFunc = (s: String) => prefix + " : " + s
}


// Add the JAR file as a dependency.
session.addDependency("<path>/myJar.jar")

// Create a new DataFrame with one column (NAME)
// that contains the name "Raymond".
val myDf = session.sql("select 'Raymond' NAME")

// Register the function that you defined earlier as an anonymous UDF.
val readFileUdf = udf(UDFCode.readFileFunc)

// Call UDF for the values in the NAME column of the DataFrame.
myDf.withColumn("CONCAT", readFileUdf(col("NAME"))).show()

10.創(chuàng)建UDTF(User-Defined Table Functions)

要在Snowpark中創(chuàng)建和注冊UDTF，你需要執(zhí)行以下步驟：
1.為UDTF定義一個類
2.創(chuàng)建UDTF實例并注冊

定義UDTF類

定義一個類，該類繼承自com.snowflake.snowpark.udtf包中的UDTFn類之一（例如UDTF0、UDTF1等），其中n表示您的UDTF的輸入?yún)?shù)數(shù)量。例如，如果您的UDTF傳入2個輸入?yún)?shù)，請繼承UDTF2類。

在您的類中，重寫以下方法：

  • outputSchema()：返回一個types.StructType對象，描述返回行中字段的名稱和類型（即輸出的“模式”）。
  • process()：對輸入分區(qū)中的每一行調(diào)用一次此方法（請參閱下面的注意事項）。
  • endPartition()：在所有行都傳遞給process()后，對每個分區(qū)調(diào)用一次此方法。

當調(diào)用UDTF時，行在傳遞給UDTF之前被分組到分區(qū)中：

  • 如果調(diào)用UDTF的語句指定了PARTITION子句（顯式分區(qū)），則該子句確定如何對行進行分區(qū)。
  • 如果語句沒有指定PARTITION子句（隱式分區(qū)），Snowflake將確定最佳的行分區(qū)方式。

Table Functions and Partitions

在將行傳遞給表函數(shù)之前，這些行可以被分組到分區(qū)中。分區(qū)有兩個主要優(yōu)勢：

  1. 分區(qū)允許Snowflake將工作負載分割成多個部分，以提高并行性和性能。
  2. 分區(qū)允許Snowflake將具有相同特征的所有行作為一個組進行處理。您可以基于組中的所有行返回結(jié)果，而不僅僅是針對單個行。

例如，您可以將股票價格數(shù)據(jù)按照每只股票分組為一個組?？梢詫⑼还镜乃泄善眱r格一起分析，同時可以獨立地分析每個公司的股票價格，而不受其他公司的影響。

數(shù)據(jù)可以顯式分區(qū)或隱式分區(qū)。

顯式分區(qū)

partitioning into multiple groups

SELECT *
    FROM stocks_table AS st,
         TABLE(my_udtf(st.symbol, st.transaction_date, st.price) OVER (PARTITION BY st.symbol))

partitioning into single group

SELECT *
    FROM stocks_table AS st,
         TABLE(my_udtf(st.symbol, st.transaction_date, st.price) OVER (PARTITION BY 1))
隱式分區(qū)

如果表函數(shù)沒有使用PARTITION BY子句顯式地對行進行分區(qū)，那么Snowflake通常會隱式地對行進行分區(qū)，以利用并行處理來提高性能。

分區(qū)的數(shù)量通常基于諸如處理函數(shù)的倉庫大小和輸入關(guān)系的基數(shù)等因素。行通常根據(jù)行的物理位置（例如，按micro-partition）分配到特定的分區(qū)，因此分區(qū)分組沒有意義。

重寫outputSchema方法

def outputSchema(): StructType

例如，如果您的UDTF返回一個只有一個整數(shù)字段的行：

override def outputSchema(): StructType = StructType(StructField("C1", IntegerType))

重寫process方法

def process(arg0: A0, ... arg<n> A<n>): Iterable[Row]

此方法對輸入分區(qū)中的每一行調(diào)用一次。

在process()方法中，構(gòu)建并返回一個包含要由UDTF返回的數(shù)據(jù)的Row對象的Iterable。行中的字段必須使用您在outputSchema方法中指定的類型。
例如，如果您的UDTF生成行，則為生成的行構(gòu)建并返回一個Row對象的Iterable：

override def process(start: Int, count: Int): Iterable[Row] =
    (start until (start + count)).map(Row(_))

重寫endPartition方法

覆蓋endPartition方法并添加在所有輸入分區(qū)的行都已傳遞給process方法后執(zhí)行的代碼。endPartition方法對于每個輸入分區(qū)都會被調(diào)用一次。

def endPartition(): Iterable[Row]

如果您在每個分區(qū)的末尾不需要返回額外的行，請返回一個空的 Row 對象的可迭代對象。例如：

override def endPartition(): Iterable[Row] = Array.empty[Row]

UDTF example

class MyRangeUdtf extends UDTF2[Int, Int] {
  override def process(start: Int, count: Int): Iterable[Row] =
    (start until (start + count)).map(Row(_))
  override def endPartition(): Iterable[Row] = Array.empty[Row]
  override def outputSchema(): StructType = StructType(StructField("C1", IntegerType))
}

注冊UDTF

接下來，創(chuàng)建一個新類的實例，并通過調(diào)用 UDTFRegistration 的方法來注冊該類。您可以注冊一個臨時的或永久的 UDTF。

注冊Temporary UDTF
// Register the MyRangeUdtf class that was defined in the previous example.
val tableFunction = session.udtf.registerTemporary(new MyRangeUdtf())
// Use the returned TableFunction object to call the UDTF.
session.tableFunction(tableFunction, lit(10), lit(5)).show

// Register the MyRangeUdtf class that was defined in the previous example.
val tableFunction = session.udtf.registerTemporary("myUdtf", new MyRangeUdtf())
// Call the UDTF by name.
session.tableFunction(TableFunction("myUdtf"), lit(10), lit(5)).show
注冊Permanent UDTF
// Register the MyRangeUdtf class that was defined in the previous example.
val tableFunction = session.udtf.registerPermanent("myUdtf", new MyRangeUdtf(), "@mystage")
// Call the UDTF by name.
session.tableFunction(TableFunction("myUdtf"), lit(10), lit(5)).show
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 一.Hadoop 1.hdfs寫流程 2.hdfs讀流程 3.hdfs的體系結(jié)構(gòu) 4.一個datanode 宕機,...
    qydong閱讀 2,549評論 0贊 0
  • 此篇內(nèi)容：hive自定義函數(shù)UDF、UDTF，壓縮存儲方式，hive優(yōu)化、hive實際編程基本涵蓋了hive基礎(chǔ)及...
    大數(shù)據(jù)階梯之路閱讀 902評論 0贊 4
  • 30.域的含義 域英文叫DOMAIN——域(Domain)是Windows網(wǎng)絡(luò)中獨立運行的單位，域之間相互訪問則需...
    saluch閱讀 156評論 0贊 0
  • 一. Flink 1.9 新特性 總的變更：https://issues.apache.org/jira/secu...
    sj_91d7閱讀 3,295評論 0贊 2
  • 一、Hive基本概念 Hive 是基于 Hadoop 的一個數(shù)據(jù)倉庫工具，可以將結(jié)構(gòu)化的數(shù)據(jù)文件映射為一張表，并提...
    CJ21閱讀 1,877評論 0贊 13

友情鏈接更多精彩內(nèi)容