Snowpark Connect for Spark 호환성 가이드¶

지원되지 않는 데이터 타입¶

• DayTimeIntervalType

• YearMonthIntervalType

• UserDefinedTypes

암시적 데이터 타입 변환¶

Snowpark Connect for Spark 사용 시 데이터 타입이 처리되는 방식에 유의하세요. Snowpark Connect for Spark 는 ByteType, ShortType, IntegerType`을 :code:`LongType`으로 암시적으로 표현합니다. 즉, 열이나 데이터를 :code:`ByteType, ShortType 또는 IntegerType`으로 정의할 수는 있지만 이 데이터는 |spconnect| 에서 :code:`LongType`으로 표현되고 반환됩니다.  마찬가지로, 특정 작업과 컨텍스트에 따라 :code:`FloatType 및 DoubleType`에 대해서도 암시적 변환이 발생할 수 있습니다. Snowflake 실행 엔진은 데이터 타입 압축을 내부적으로 처리하며 실제로 데이터를 :code:`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)

:code:`NullType`의 미묘한 차이¶

Snowpark Connect for Spark 는 Spark에서 지원되는 `NullType<https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.types.NullType.html>`_ 데이터 타입을 지원하지 않습니다. 이로 인해 데이터 프레임에서 Null 또는 :code:`None`을 사용할 때 동작이 변경됩니다.

Spark에서 NULL 리터럴(예: lit(None))은 :code:`NullType`으로 자동 추론됩니다. Snowpark Connect for Spark 에서는 스키마 추론 중 :code:`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, :code:`ObjectType`의 정형화 데이터 타입¶

Snowpark Connect for Spark 에서는 기본적으로 정형화 타입 지원을 사용할 수 없지만, ARRAY, MAP, Object 데이터 타입은 유형이 지정되지 않은 일반 컬렉션으로 처리됩니다. 즉, 정형화 타입 지원에서 제공하는 것과 달리 요소 유형, 필드 이름, 스키마 또는 null 허용 여부가 적용되지 않습니다.

이 지원에 대한 종속성이 있는 경우 계정 팀과 협업하여 계정에서 이 기능을 활성화하세요.

:code:`StructType`의 차이점¶

Spark는 사용자 정의 함수(UDF)에 사용할 StructType`을 변환할 때 Python의 :code:`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에 포함된 이러한 파일에서 함수를 가져올 수 있습니다. 이 종속성 처리 동작은 :ref:`label-udf_python_stage`에 설명된 대로 작동합니다.

외부 라이브러리 및 파일을 포함하려면 파일에 대한 스테이지 경로를 구성 설정 :code:`snowpark.connect.udf.imports`의 값으로 제공합니다. 구성 값은 파일에 대한 스테이지 경로의 배열이어야 하며, 경로는 쉼표로 구분됩니다.

다음 예제의 코드에는 UDF의 실행 컨텍스트에 두 개의 파일이 포함되어 있습니다. 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 Catalog 공급자 지원¶

• Snowflake만 카탈로그 공급자로 지원됩니다.

APIs 카탈로그는 지원되지 않습니다.¶

• registerFunction

• listFunctions

• getFunction

• functionExists

• createExternalTable

APIs 카탈로그는 일부 지원됩니다.¶

• :code:`createTable`(외부 테이블 미지원)

Snowflake 관리형 Iceberg 테이블¶

Spark용 Snowpark Connect는 외부 관리형 Apache Iceberg™ 테이블(Iceberg 테이블 및 카탈로그 연결 데이터베이스 포함)과 함께 작동합니다.

외부 관리형 Iceberg 테이블¶