Snowpark Connect for Spark 互換ガイド¶

サポートされていないデータ型¶

• DayTimeIntervalType

• YearMonthIntervalType

• UserDefinedTypes

暗黙的なデータ型変換¶

Snowpark Connect for Spark を使用する場合は、データ型の処理方法に注意してください。Snowpark Connect for Spark は暗黙的に LongType として ByteType、 ShortType`および :code:`IntegerType を表します。 これは、 ByteType、 ShortType または IntegerType`で列やデータを定義することができる場合、データは :code:`LongType として Snowpark Connect for Spark によって表示され、返されることを意味します。同様に、特定の操作とコンテキストに応じた暗黙的な変換が FloatType および DoubleType で起こることがあります。Snowflake 実行エンジンは内部的にデータ型の圧縮を処理し、実際にデータを Byte または :code:`Short`として保存することができます。ただし、これらは実装の詳細とみなされ、エンドユーザーに公開されません。

意味的には、この表現は Spark クエリの正確さに影響を与えません。

次の例は、Sparkと Snowpark Connect for Spark のクエリ結果でのデータ型処理方法の違いを示しています。

query = """
    SELECT * FROM VALUES
    (float(1.0), double(1.0), 1.0, "1", true, :code:`NULL`),
    (float(2.0), double(2.0), 2.0, "2", false, :code:`NULL`),
    (float(3.0), double(3.0), :code:`NULL`, "3", false, :code:`NULL`)
    AS tab(a, b, c, d, e, f)
    """

spark.sql(query).printSchema()

root
 |-- a: float (nullable = false)
 |-- b: double (nullable = false)
 |-- c: decimal(2,1) (nullable = true)
 |-- d: string (nullable = false)
 |-- e: boolean (nullable = false)
 |-- f: void (nullable = true)

snowpark_connect_spark.sql(query).printSchema()

root
 |-- a: double (nullable = false)
 |-- b: double (nullable = false)
 |-- c: decimal (nullable = true)
 |-- d: string (nullable = false)
 |-- e: boolean (nullable = true)
 |-- f: string (nullable = true)

NullType ニュアンス¶

Snowpark Connect for Spark は、Sparkでサポートされている NullType データ型をサポートしていません。これにより、データフレームの Null や None を使用すると、動作が変更されます。

Sparkでは、リテラル NULL （例: lit(None)）は、 NullType として自動的に推測されます。|spconnect|内では、スキーマ推論中は StringType として推測されます。

df = self.spark.range(1).select(lit(None).alias("null_col"))
field = df.schema["null_col"]

# Spark: StructField('null_col', :code:`NullType`(), True)
# Snowpark Connect for Spark: StructField('null_col', :code:`StringType`(), True)

ArrayType MapType および ObjectType の構造化データ型。¶

構造化型のサポートは Snowpark Connect for Spark ではデフォルトでは利用できませんが、 ARRAY、 MAP および Object データ型は一般的な型付けされていないコレクションとして扱われます。これは、構造化型のサポートによって提供されるものとは異なり、要素型、フィールド名、スキーマ、またはNULL値の許容が強制されないことを意味します。

このサポートと依存関係がある場合は、アカウントチームと協力して、アカウントでこの機能を有効にしてください。

StructType の相違¶

Sparkがユーザー定義関数（UDF）に使用するため StructType を変換するとき、Python の tuple 型に変換します。Snowpark Connect for Spark は StructType を Pythonの:code:dict 型に変換します。これには、要素のアクセスと出力に基本的な違いがあります。

• Sparkは0、1、2、3などのインデックスにアクセスします。

• Snowpark Connect for Spark は「_1」、「_2」などを使用してインデックスにアクセスします。

def f(e):
    return e[0]

    df = self.spark.createDataFrame([((1.0, 1.0), (1, 1))], ["c1", "c2"])
    result = df.select("*", udf(f, DoubleType())("c1"))

# This results in an index access issue. Workaround is to use _1, _2 as indicies.
# Workaround:

def f(e):
    return e['_1']

row = (
    self.spark.range(1)
    .selectExpr("struct(1, 2) as struct")
    .select(
        udf(lambda x: x, "struct<col1:int,col2:int>")("struct"),
    )
    .first()
)

self.assertEquals(row[0], Row(col1=1, col2=2))

# Spark: Row(col1=1, col2=2)

# Snowpark Connect for Spark: {'col1': 1, 'col2': 2}

反復子型の UDFs¶

反復子は、戻り型または入力型としてサポートされていません。

# This will not work
def func(iterator):
  for _ in iterator:
              ...

df = self.spark.range(10)
actual = df.repartition(1).mapInArrow(func, "a long").collect()

Importing files to a Python UDF¶

With Snowpark Connect for Spark, you can specify external libraries and files in Python UDFs. Snowflake includes Python files and archives in your code's execution context. You can import functions from these included files in a UDF without additional steps. This dependency-handling behavior works as described in ステージからアップロードされたコードを使用したPython UDF の作成.

To include external libraries and files, you provide stage paths to the files as the value of the configuration setting snowpark.connect.udf.imports. The configuration value should be an array of stage paths to the files, where the paths are separated by commas.

Code in the following example includes two files in the UDF's execution context. The UDF imports functions from these files and uses them in its logic.

# Files need to be previously staged
spark.conf.set("snowpark.connect.udf.imports", "[@stage/library.py, @other_lib.zip]")

@udf(returnType = StringType())
def import_example(input: str) -> str:
  from library import first_function
  from other_lib.custom import second_function

  return first_function(input) + second_function(input)

spark.range(1).select(import_read_example("example_string")).show()

You can use the snowpark.connect.udf.imports setting to include other kinds of files as well, such as those with data your code needs to read. Note that when you do this, your code should only read from the included files; any writes to such files will be lost after the function's execution ends.

# Files need to be previously staged
spark.conf.set("snowpark.connect.udf.imports", "[@stage/data.csv]")

@udf(returnType = StringType())
def import_read_example(file_name: str) -> str:
  with open(file_name) as f:
    return f.read()

spark.range(1).select(import_read_example("data.csv")).show()

Snowflake Horizon カタログ プロバイダーのサポート¶

• カタログ プロバイダーとしてサポートされているのはSnowflakeのみです。

サポートされていないカタログ APIs¶

• registerFunction

• listFunctions

• getFunction

• functionExists

• createExternalTable

部分的にサポートされているカタログ APIs¶

• createTable （外部テーブルのサポートなし）

Snowflake 管理 Iceberg テーブル¶

Spark 用 Snowpark Connectは、外部で管理された Iceberg テーブルやカタログリンクのデータベースを含む Apache Iceberg™テーブルと動作します。

外部管理 Iceberg テーブル¶