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 として自動的に推測されます。Snowpark Connect for Spark 内では、スキーマ推論中は 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()

Python UDF へのファイルのインポート¶

Snowpark Connect for Spark を使用すると、Python UDFs で外部ライブラリとファイルを指定できます。Snowflakeには、コードの実行コンテキストにPythonファイルとアーカイブが含まれています。追加の手順なしで、これらの UDF に含まれるファイルから関数をインポートできます。この依存関係の動作は、 ステージからアップロードされたコードを使用したPython UDF の作成 で説明されているように機能します。

外部ライブラリとファイルを含めるには、 snowpark.connect.udf.imports 構成設定の値としてファイルへのステージパスを指定します。構成値は、ファイルへのステージパスの配列である必要があります。パスはコンマで区切られます。

次の例のコードには、 UDF の実行コンテキストの2つのファイルが含まれています。UDF はこれらのファイルから関数をインポートし、ロジックで使用します。

# 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()

snowpark.connect.udf.imports 設定を使用して、コードが読み取る必要のあるデータなど、他の種類のファイルも含めます。これを行う場合、コードは含まれているファイルからのみ読み取る必要があることに注意してください。関数の実行が終了すると、そのようなファイルへの書き込みは失われます。

# 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 テーブル¶