Snowpark Migration Accelerator: Python용 코드 발행

SPRKPY1089

메시지: The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.

카테고리: 경고

설명

DataFrameWriter .options 값이 Snowpark에서 다를 수 있으므로 동작이 올바른지 확인하기 위해 유효성 검사가 필요할 수 있습니다.

시나리오

옵션의 지원 여부 또는 파일 작성에 사용되는 형식에 따라 몇 가지 시나리오가 있습니다.

시나리오 1

입력

다음은 현재 지원 되는 옵션인 sep 및 nullValue 를 추가한 메서드 옵션의 사용 예입니다.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])

df.write.options(nullValue="myVal", sep=",").csv("some_path")
Copy

출력

이 도구는 필수 유효성 검사임을 나타내는 EWI SPRKPY1089 를 추가합니다.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])
#EWI: SPRKPY1089 => The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.
df.write.options(nullValue="myVal", sep=",").csv("some_path")
Copy

권장 수정

Snowpark API 는 이러한 매개 변수를 지원하므로 마이그레이션 후 동작을 확인하는 작업만 하면 됩니다. 지원되는 매개 변수를 확인하려면 동등성 테이블 을 참조하십시오.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])
#EWI: SPRKPY1089 => The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.
df.write.options(nullValue="myVal", sep=",").csv("some_path")
Copy
시나리오 2

입력

이 시나리오는 옵션 사용법을 보여 주지만 지원되지 않는 헤더 옵션을 추가합니다.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])

df.write.options(header=True, sep=",").csv("some_path")
Copy

출력

이 도구는 유효성 검사가 필요함을 나타내는 EWI SPRKPY1089 를 추가합니다.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])
#EWI: SPRKPY1089 => The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.
df.write.options(header=True, sep=",").csv("some_path")
Copy

권장 수정

이 시나리오의 경우 필요에 따라 변경할 수 있는지 확인하기 위해 Snowpark 형식 유형 옵션 을 평가하는 것이 좋습니다. 또한 변경 후 동작을 확인하십시오.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])
#EWI: SPRKPY1089 => The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.
df.write.csv("some_path")
Copy
시나리오 3

입력

이 시나리오에서는 JSON 메서드를 사용하는 지원되는 sep 옵션을 추가합니다.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])

df.write.options(nullValue="myVal", sep=",").json("some_path")
Copy

출력

이 도구는 유효성 검사가 필요함을 나타내는 EWI SPRKPY1089 를 추가합니다.

  • 참고: 이 시나리오는 PARQUET 에도 적용됩니다.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])
#EWI: SPRKPY1089 => The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.
df.write.options(nullValue="myVal", sep=",").json("some_path")
Copy

권장 수정

파일 형식 JSON 은 sep 매개 변수를 지원하지 않으므로 필요에 따라 변경할 수 있는지 확인하기 위해 Snowpark 형식 유형 옵션 을 평가하는 것이 좋습니다. 또한 변경 후 동작을 확인하십시오.

df = spark.createDataFrame([(1, "myVal")], [2, "myVal2"], [None, "myVal3" ])
#EWI: SPRKPY1089 => The pyspark.sql.readwriter.DataFrameWriter.options values in Snowpark may be different, so required validation might be needed.
df.write.json("some_path")
Copy

추가 권장 사항

  • 일부 지원되지 않는 매개 변수가 있으므로 동등 매개 변수 테이블 을 확인하여 변환 후 동작을 확인하는 것이 좋습니다.

  • 동등 매개 변수 테이블:

Snowpark는 일부 매개 변수에 대해 동등 매개 변수 목록을 지원할 수 있습니다.

PySpark 옵션

SnowFlake 옵션

지원되는 파일 형식

설명

SEP

FIELD_DELIMITER

CSV

입력 파일에서 필드를 구분하는 1개 이상의 단일 바이트 또는 멀티바이트 문자입니다.

LINESEP

RECORD_DELIMITER

CSV

입력 파일에서 레코드를 구분하는 하나 이상의 문자입니다.

QUOTE

FIELD_OPTIONALLY_ENCLOSED_BY

CSV

문자열을 묶는 데 사용되는 문자입니다.

NULLVALUE

NULL_IF

CSV

SQL NULL로 변환하거나 그 반대로 변환하는 데 사용되는 문자열입니다.

DATEFORMAT

DATE_FORMAT

CSV

로딩할 데이터 파일의 날짜 값 형식을 정의하는 문자열입니다.

TIMESTAMPFORMAT

TIMESTAMP_FORMAT

CSV

로딩할 데이터 파일의 타임스탬프 값 형식을 정의하는 문자열입니다.

사용되는 매개 변수가 목록에 없으면 API 가 오류를 throw합니다.

SPRKPY1088

메시지: The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.

카테고리: 경고

설명

DataFrameWriter.option 값이 Snowpark에서 다를 수 있으므로 동작이 올바른지 확인하기 위해 유효성 검사가 필요할 수 있습니다.

시나리오

지원되는 옵션 또는 파일 작성에 사용되는 형식에 따라 몇 가지 시나리오가 있습니다.

시나리오 1

입력

다음은 현재 지원되는 sep 옵션을 추가하는 메서드 옵션의 사용 예시입니다.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])

df.write.option("sep", ",").csv("some_path")
Copy

출력

이 도구는 필수 유효성 검사임을 나타내는 EWI SPRKPY1088 을 추가합니다.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])
#EWI: SPRKPY1088 => The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.
df.write.option("sep", ",").csv("some_path")
Copy

권장 수정

Snowpark API 는 이 매개 변수를 지원하므로 마이그레이션 후 동작을 확인하는 작업만 수행하면 됩니다. 지원되는 매개 변수를 확인하려면 동등성 테이블 을 참조하십시오.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])
#EWI: SPRKPY1088 => The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.
df.write.option("sep", ",").csv("some_path")
Copy
시나리오 2

입력

이 시나리오는 옵션 사용법을 보여 주지만 지원되지 않는 헤더 옵션을 추가합니다.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])

df.write.option("header", True).csv("some_path")
Copy

출력

이 도구는 유효성 검사가 필요함을 나타내는 EWI SPRKPY1088 을 추가합니다.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])
#EWI: SPRKPY1088 => The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.
df.write.option("header", True).csv("some_path")
Copy

권장 수정

이 시나리오의 경우 필요에 따라 변경할 수 있는지 확인하기 위해 Snowpark 형식 유형 옵션 을 평가하는 것이 좋습니다. 또한 변경 후 동작을 확인하십시오.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])
#EWI: SPRKPY1088 => The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.
df.write.csv("some_path")
Copy
시나리오 3

입력

이 시나리오에서는 JSON 메서드를 사용하는 지원되는 sep 옵션을 추가합니다.

  • 참고: 이 시나리오는 PARQUET 에도 적용됩니다.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])

df.write.option("sep", ",").json("some_path")
Copy

출력

이 도구는 유효성 검사가 필요함을 나타내는 EWI SPRKPY1088 을 추가합니다.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])
#EWI: SPRKPY1088 => The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.
df.write.option("sep", ",").json("some_path")
Copy

권장 수정

파일 형식 JSON 은 sep 매개 변수를 지원하지 않으므로 필요에 따라 변경할 수 있는지 확인하기 위해 Snowpark 형식 유형 옵션 을 평가하는 것이 좋습니다. 또한 변경 후 동작을 확인하십시오.

df = spark.createDataFrame([(100, "myVal")], ["ID", "Value"])
#EWI: SPRKPY1088 => The pyspark.sql.readwriter.DataFrameWriter.option values in Snowpark may be different, so required validation might be needed.
df.write.json("some_path")
Copy

추가 권장 사항

  • 일부 지원되지 않는 매개 변수가 있으므로 동등 매개 변수 테이블 을 확인하여 변환 후 동작을 확인하는 것이 좋습니다.

  • 동등 매개 변수 테이블:

PySpark 옵션

SnowFlake 옵션

지원되는 파일 형식

설명

SEP

FIELD_DELIMITER

CSV

입력 파일에서 필드를 구분하는 1개 이상의 단일 바이트 또는 멀티바이트 문자입니다.

LINESEP

RECORD_DELIMITER

CSV

입력 파일에서 레코드를 구분하는 하나 이상의 문자입니다.

QUOTE

FIELD_OPTIONALLY_ENCLOSED_BY

CSV

문자열을 묶는 데 사용되는 문자입니다.

NULLVALUE

NULL_IF

CSV

SQL NULL로 변환하거나 그 반대로 변환하는 데 사용되는 문자열입니다.

DATEFORMAT

DATE_FORMAT

CSV

로딩할 데이터 파일의 날짜 값 형식을 정의하는 문자열입니다.

TIMESTAMPFORMAT

TIMESTAMP_FORMAT

CSV

로딩할 데이터 파일의 타임스탬프 값 형식을 정의하는 문자열입니다.

사용되는 매개 변수가 목록에 없으면 API 가 오류를 throw합니다.

SPRKPY1011

메시지: pyspark.sql.dataframe.DataFrameStatFunctions.approxQuantile에는 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.dataframe.DataFrameStatFunctions.approxQuantile 의 사용을 감지했을 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

Pyspark는 두 가지 다른 approxQuantile 함수를 사용한다는 것을 이해하는 것이 중요합니다. 여기서는 DataFrameStatFunctions approxQuantile 버전을 사용합니다.

import tempfile
from pyspark.sql import SparkSession, DataFrameStatFunctions
spark = SparkSession.builder.getOrCreate()
data = [['Q1', 300000],
        ['Q2', 60000],
        ['Q3', 500002],
        ['Q4', 130000]]

columns = ['Quarter', 'Gain']
df = spark.createDataFrame(data, columns)
aprox_quantille = DataFrameStatFunctions(df).approxQuantile('Gain', [0.25, 0.5, 0.75], 0)
print(aprox_quantille)
Copy

출력

SMA 는 접근 사분위수가 사용되는 라인 위에 EWI SPRKPY1011 을 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

import tempfile
from snowflake.snowpark import Session, DataFrameStatFunctions
spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 300000],
        ['Q2', 60000],
        ['Q3', 500002],
        ['Q4', 130000]]

columns = ['Quarter', 'Gain']
df = spark.createDataFrame(data, columns)
#EWI: SPRKPY1011 => pyspark.sql.dataframe.DataFrameStatFunctions.approxQuantile has a workaround, see documentation for more info
aprox_quantille = DataFrameStatFunctions(df).approxQuantile('Gain', [0.25, 0.5, 0.75], 0)
Copy

권장 수정

Snowpark approxQuantile 메서드를 사용할 수 있습니다. 일부 매개 변수가 일치하지 않으므로 일부 수동 조정이 필요합니다. 출력 코드 예제의 경우 권장 수정 사항이 될 수 있습니다.

from snowflake.snowpark import Session # remove DataFrameStatFunctions because is not required
...
df = spark.createDataFrame(data, columns)

aprox_quantille = df.stat.approx_quantile('Ammount', [0.25, 0.5, 0.75])
Copy

SnowPark 에는 DataFrame.approxQuantile의 상대 오류 매개 변수가 존재하지 않습니다.

추가 권장 사항

SPRKPY1040

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.explode로 해결할 수 있습니다. 자세한 내용은 설명서를 참조하십시오.

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.explode 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

EWI 를 생성하는 pyspark.sql.functions.explode 함수의 사용 예제는 다음과 같습니다. 이 예제에서는 explode 함수를 사용하여 숫자 열에 대한 배열 항목당 하나의 행을 생성합니다.

df = spark.createDataFrame([("Alice", [1, 2, 3]), ("Bob", [4, 5]), ("Charlie", [6, 7, 8, 9])], ["name", "numbers"])
exploded_df = df.select("name", explode(df.numbers).alias("number"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1040 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([("Alice", [1, 2, 3]), ("Bob", [4, 5]), ("Charlie", [6, 7, 8, 9])], ["name", "numbers"])
#EWI: SPRKPY1040 => pyspark.sql.functions.explode has a workaround, see documentation for more info
exploded_df = df.select("name", explode(df.numbers).alias("number"))
Copy

권장 수정

해결 방법으로 explode 함수에 대한 확장자를 제공하는 snowpark_extensions 패키지를 가져오면 됩니다.

import snowpark_extensions

df = spark.createDataFrame([("Alice", [1, 2, 3]), ("Bob", [4, 5]), ("Charlie", [6, 7, 8, 9])], ["name", "numbers"])
exploded_df = df.select("name", explode(df.numbers).alias("number"))
Copy

추가 권장 사항

SPRKPY1074

메시지: 파일에 들여쓰기(공백과 탭)가 섞여 있습니다.

카테고리: 구문 분석 오류.

설명

이 문제는 도구에서 파일에 들여쓰기가 혼합되어 있는 것을 감지할 때 표시됩니다. 즉, 파일에는 공백과 탭을 조합하여 코드 라인을 들여쓰기할 수 있습니다.

시나리오

입력

Pyspark에서는 공백과 탭을 혼합하여 ID 레벨을 지정할 수 있습니다.

def foo():
    x = 5 # spaces
    y = 6 # tab
Copy

출력

SMA 는 혼합 들여쓰기 마커를 처리할 수 없습니다. 이 항목이 Python 코드 파일 SMA 에서 감지되면 첫 라인에 EWI SPRKPY1074 를 추가합니다.

## EWI: SPRKPY1074 => File has mixed indentation (spaces and tabs).
## This file was not converted, so it is expected to still have references to the Spark API
def foo():
    x = 5 # spaces
    y = 6 # tabs
Copy

권장 수정

해결책은 모든 들여쓰기 기호를 동일하게 생성하는 것입니다.

def foo():
  x = 5 # tab
  y = 6 # tab
Copy

추가 권장 사항

SPRKPY1025

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.ntile에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.ntile 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.functions.ntile 함수의 사용 예제입니다. 이 예제에서는 ntile 함수를 사용하여 행을 3개의 버킷으로 나눕니다.

df = spark.createDataFrame([("Alice", 50), ("Bob", 30), ("Charlie", 60), ("David", 90), ("Eve", 70), ("Frank", 40)], ["name", "score"])
windowSpec = Window.orderBy("score")
df_with_ntile = df.withColumn("bucket", ntile(3).over(windowSpec))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1025 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([("Alice", 50), ("Bob", 30), ("Charlie", 60), ("David", 90), ("Eve", 70), ("Frank", 40)], ["name", "score"])
windowSpec = Window.orderBy("score")
#EWI: SPRKPY1025 => pyspark.sql.functions.ntile has a workaround, see documentation for more info
df_with_ntile = df.withColumn("bucket", ntile(3).over(windowSpec))
Copy

권장 수정

Snowpark에는 이와 동등한 ntile 함수가 있지만, 이 함수에 전달되는 인자는 열이어야 합니다. 해결 방법으로 Snowflake.snowpark.functions.lit 함수를 사용하여 리터럴 인자를 열로 변환할 수 있습니다.

df = spark.createDataFrame([("Alice", 50), ("Bob", 30), ("Charlie", 60), ("David", 90), ("Eve", 70), ("Frank", 40)], ["name", "score"])
windowSpec = Window.orderBy("score")
df_with_ntile = df.withColumn("bucket", ntile(lit(3)).over(windowSpec))
Copy

추가 권장 사항

SPRKPY1087

메시지: DataFrame.writeTo 함수는 지원되지 않지만 해결 방법이 있습니다.

카테고리: 경고.

설명

pyspark.sql.dataframe.DataFrame.writeTo 함수는 지원되지 않습니다. 해결 방법은 대신 Snowpark DataFrameWriter SaveAsTable 메서드를 사용하는 것입니다.

시나리오

입력

다음은 pyspark.sql.dataframe.DataFrame.writeTo 함수를 사용하여 데이터프레임 df 를 테이블 이름 Personal_info 에 작성한 예제입니다.

df = spark.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]],
                                 schema=["FIRST_NAME", "LAST_NAME"])

df.writeTo("Personal_info")
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1087 을 추가하여 이 함수가 Snowpark에서 지원되지 않음을 알려줍니다.

df = spark.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]],
                                 schema=["FIRST_NAME", "LAST_NAME"])

#EWI: SPRKPY1087 => pyspark.sql.dataframe.DataFrame.writeTo is not supported, but it has a workaround.
df.writeTo("Personal_info")
Copy

권장 수정

해결 방법은 대신 Snowpark DataFrameWriter SaveAsTable 메서드를 사용하는 것입니다.

df = spark.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]],
                                 schema=["FIRST_NAME", "LAST_NAME"])

df.write.saveAsTable("Personal_info")
Copy

추가 권장 사항

SPRKPY1035

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.reverse에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.reverse 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.functions.reverse 함수의 사용 예제입니다. 이 예제에서 reverse 함수는 word 열의 각 문자열을 반전시키는 데 사용됩니다.

df = spark.createDataFrame([("hello",), ("world",)], ["word"])
df_reversed = df.withColumn("reversed_word", reverse(df["word"]))
df_reversed = df.withColumn("reversed_word", reverse("word"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1035 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([("hello",), ("world",)], ["word"])
#EWI: SPRKPY1035 => pyspark.sql.functions.reverse has a workaround, see documentation for more info
df_reversed = df.withColumn("reversed_word", reverse(df["word"]))
#EWI: SPRKPY1035 => pyspark.sql.functions.reverse has a workaround, see documentation for more info
df_reversed = df.withColumn("reversed_word", reverse("word"))
Copy

권장 수정

해결 방법으로 reverse 함수에 대한 확장자를 제공하는 snowpark_extensions 패키지를 가져오면 됩니다.

import snowpark_extensions

df = spark.createDataFrame([("hello",), ("world",)], ["word"])
df_reversed = df.withColumn("reversed_word", reverse(df["word"]))
df_reversed = df.withColumn("reversed_word", reverse("word"))
Copy

추가 권장 사항

SPRKPY1064

메시지: Snowflake가 대신 Snowpipe 메커니즘을 사용하므로 _ Spark 요소 _는 적용되지 않습니다.

카테고리: 경고

설명

이 문제는 도구가 pyspark.streaming 라이브러리에서 어떤 요소의 사용을 감지할 때 표시됩니다.

시나리오

입력

아래는 이 EWI 를 트리거하는 요소 중 하나가 포함된 예시입니다.

from pyspark.streaming.listener import StreamingListener

var = StreamingListener.Java
var.mro()

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])
df.show()
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1064 를 추가하여 이 함수가 적용되지 않음을 알립니다.

#EWI: SPRKPY1064 => The element does not apply since snowflake uses snowpipe mechanism instead.

var = StreamingListener.Java
var.mro()

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])
df.show()
Copy

권장 수정

SMA 는 가져오기 문을 제거하고 _Issues.csv_인벤토리에 문제를 추가하고 Spark 요소의 모든 사용처를 제거합니다.

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])
df.show()
Copy

추가 권장 사항

  • 현재 시나리오에 적용되는 방법은 Snowpipe 설명서를 확인하십시오.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1050

메시지: pyspark.conf.SparkConf.set에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.conf.SparkConf.set 의 사용을 감지할 때 발생하며, 해결이 가능합니다.

시나리오

입력

다음은 conf.set 을 사용하여 변수를 설정하는 예제입니다.

conf = SparkConf().setAppName('my_app')

conf.set("spark.storage.memoryFraction", "0.5")
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1050 을 추가합니다.

conf = SparkConf().setAppName('my_app')

#EWI: SPRKPY1050 => pyspark.conf.SparkConf.set has a workaround, see documentation for more info
conf.set("spark.storage.memoryFraction", "0.5")
Copy

권장 수정

SparkConf.set은 Pyspark에서만 사용하는 구성 설정을 설정하는 데 사용되며 Snowpark에는 적용되지 않습니다. 코드를 제거하거나 설명할 수 있습니다

#conf.set("spark.storage.memoryFraction", "0.5")
Copy

추가 권장 사항

SPRKPY1001

메시지 : 이 코드 섹션에 구문 분석 오류가 있습니다

카테고리 : 구문 분석 오류입니다.

설명

구문 분석 오류는 파일의 코드를 올바르게 읽거나 이해할 수 없는 경우(파일을 올바르게 “구문 분석”할 수 없는 경우) Snowpark Migration Accelerator(SMA)에서 보고합니다. 이 문제 코드는 파일에 1개 이상의 구문 분석 오류가 있을 때 표시됩니다.

시나리오 <a href=”#scenario”id=”scenario”>

입력: 예를 들어, 코드에 잘못된 구문이 있는 경우 EWI 메시지가 표시됩니다.

def foo():
    x = %%%%%%1###1
Copy

출력: SMA 구문 분석 오류를 찾아 해당 EWI 메시지를 추가하여 구문 분석 오류를 설명합니다.

def foo():
    x
## EWI: SPRKPY1101 => Unrecognized or invalid CODE STATEMENT @(2, 7). Last valid token was 'x' @(2, 5), failed token '=' @(2, 7)
##      = %%%%%%1###1
Copy

추가 권장 사항

  • 파일에 유효한 Python 코드가 포함되어 있는지 확인합니다. (문제.csv 파일을 사용하여 EWI 코드가 있는 모든 파일을 찾아 구문 분석 오류로 인해 도구에서 처리되지 않은 파일을 확인할 수 있습니다.) 코드의 일부만 도구에 입력되기 때문에 많은 구문 분석 오류가 발생하므로 코드가 소스에서 실행되는지 확인하는 것이 좋습니다. 유효한 경우, SMA의 문제 보고 옵션을 사용하여 구문 분석 오류가 발생했다고 보고하십시오. 이 문제를 파일로 제출할 때 설명에 구문 분석 오류를 유발한 코드 라인을 포함하십시오.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1021

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.last에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.last 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.functions.last 함수의 사용 예제입니다. 이 예제에서는 last 함수를 사용하여 각 이름에 대한 마지막 값 을 가져옵니다.

df = spark.createDataFrame([("Alice", 1), ("Bob", 2), ("Charlie", 3), ("Alice", 4), ("Bob", 5)], ["name", "value"])
df_grouped = df.groupBy("name").agg(last("value").alias("last_value"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1021 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([("Alice", 1), ("Bob", 2), ("Charlie", 3), ("Alice", 4), ("Bob", 5)], ["name", "value"])
#EWI: SPRKPY1021 => pyspark.sql.functions.last has a workaround, see documentation for more info
df_grouped = df.groupBy("name").agg(last("value").alias("last_value"))
Copy

권장 수정

해결 방법으로 Snowflake LAST_VALUE 함수를 사용할 수 있습니다. Snowpark에서 이 함수를 호출하려면 Snowflake.snowpark.functions.call_builtin 함수를 사용하고 last_value 문자열을 첫 번째 인수로, 해당 열을 두 번째 인수로 전달합니다. last 함수에서열의 이름을 사용했다면 call_builtin 함수를 호출할 때 열로 변환해야 합니다.

df = spark.createDataFrame([("Alice", 1), ("Bob", 2), ("Charlie", 3), ("Alice", 4), ("Bob", 5)], ["name", "value"])
df_grouped = df.groupBy("name").agg(call_builtin("last_value", col("value")).alias("last_value"))
Copy

추가 권장 사항

설명: >- CSV, JSON 및 PARQUET 메서드의 mode 매개 변수가 overwrite 로 변환됩니다.

SPRKPY1070

메시지: mode 인자가 overwrite 로 변환되고 변수 값을 확인한 후 해당 부울 값을 설정합니다.

카테고리: 경고

설명

다음의 사용법이 있는 경우:

이 도구는 매개 변수 mode 를 분석하여 값이 overwrite 인지 확인합니다.

시나리오

시나리오 1

입력 코드

이 시나리오에서 도구는 모드 매개 변수가 해당 부울 값을 설정할 수 있음을 감지합니다.

df.write.csv(file_path, mode="overwrite")
Copy

출력 코드:

SMA 도구는 모드 매개 변수를 분석하여 값이 overwrite 인 것을 확인하고 해당 부울 값을 설정합니다

df.write.csv(file_path, format_type_options = dict(compression = "None"), overwrite = True)
Copy

권장 수정

이 시나리오는 도구가 해당 변환을 수행했기 때문에 권장되는 수정 사항이 없습니다.

시나리오 2:

입력 코드

이 시나리오에서 도구는 overwrite 값의 유효성을 검사할 수 없습니다.

df.write.csv(file_path, mode=myVal)
Copy

출력 코드:

SMA 는 모드 변수가 ‘overwrite’로 변환되었음을 나타내는 EWI 메시지를 추가하지만, 변수 값을 확인하고 올바른 부울 값을 설정하는 것이 좋다는 것을 알려주기 위한 것이기도 합니다.

#EWI: SPRKPY1070 => The 'mode' argument is transformed to 'overwrite', check the variable value and set the corresponding bool value.
df.write.csv(file_path, format_type_options = dict(compression = "None"), overwrite = myVal)
Copy

권장 수정

mode 매개 변수의 값을 확인하고 overwrite 매개 변수에 올바른 값을 추가합니다.

df.write.csv(file_path, format_type_options = dict(compression = "None"), overwrite = True)
Copy

추가 권장 사항

SPRKPY1083

메시지: DataFrameWriter.save 함수는 지원되지 않습니다. 해결 방법은 대신 Snowpark DataFrameWriter copy_into_location 메서드를 사용하는 것입니다.

카테고리: 경고

설명

DataFrameWriter.save 함수는 지원되지 않습니다. 해결 방법은 대신 Snowpark DataFrameWriter 메서드를 사용하는 것입니다.

시나리오

이 메서드 DataFrameWriter.save(경로, 형식, 모드, partitionBy, **옵션) 에 대한 Spark 서명은 Snowpark에 존재하지 않습니다. 따라서 로딩 함수를 사용할 때마다 출력 코드에 EWI 를 사용하면 됩니다.

시나리오 1

입력 코드

다음은 CSV 형식으로 데이터를 저장하는 예제입니다.

path_csv_file = "/path/to/file.csv"

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = my_session.createDataFrame(data, schema=["Name", "Age", "City"])

df.write.save(path_csv_file, format="csv")
df.write.save(path_csv_file, format="csv", mode="overwrite")
df.write.save(path_csv_file, format="csv", mode="overwrite", lineSep="\r\n", dateFormat="YYYY/MM/DD")
df.write.save(path_csv_file, format="csv", mode="overwrite", partitionBy="City", lineSep="\r\n", dateFormat="YYYY/MM/DD")
Copy

출력 코드

이 도구는 출력 코드에 EWI SPRKPY1083 을 추가하여 이 함수가 Snowpark에서 지원되지 않지만 우회 방법이 있음을 알려줍니다.

path_csv_file = "/path/to/file.csv"

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = my_session.createDataFrame(data, schema=["Name", "Age", "City"])

#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_csv_file, format="csv")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_csv_file, format="csv", mode="overwrite")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_csv_file, format="csv", mode="overwrite", lineSep="\r\n", dateFormat="YYYY/MM/DD")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_csv_file, format="csv", mode="overwrite", partitionBy="City", lineSep="\r\n", dateFormat="YYYY/MM/DD")
Copy

권장 수정

해결 방법으로 Snowpark DataFrameWriter 메서드를 대신 사용할 수 있습니다.

  • path 및 format 매개 변수 수정:

    • load 메서드를 csv 또는 copy_into_location 메서드로 바꿉니다.

    • copy_into_location 메서드를 사용하는 경우 file_format_type 매개 변수 를 사용하여 형식을 지정해야 합니다.

    • 첫 번째 매개 변수 path 는 Snowpark 와 동등하게 생성하기 위해 스테이지에 있어야 합니다.

아래는 임시 스테이지를 생성하고 그 안에 파일을 넣은 다음 위에서 언급한 메서드 중 하나를 호출하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Stage creation

temp_stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
my_session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}').show()
my_session.file.put(f"file:///path/to/file.csv", f"@{temp_stage}")
stage_file_path = f"{temp_stage}file.csv"

## Using csv method
df.write.csv(stage_file_path)

## Using copy_into_location method
df.write.copy_into_location(stage_file_path, file_format_type="csv")
Copy

아래는 데이지 체인에 mode 메서드를 추가하고 overwrite 를 매개 변수로 추가하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Using csv method
df.write.mode("overwrite").csv(temp_stage)

## Using copy_into_location method
df.write.mode("overwrite").copy_into_location(temp_stage, file_format_type="csv")
Copy

  • partitionBy 매개 변수:

    • 다음과 같이 CSV 메서드에서 partition_by 매개 변수를 사용합니다.

다음은 CSV 메서드에서 partition_by 매개 변수를 사용한 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Using csv method
df.write.csv(temp_stage, partition_by="City")

## Using copy_into_location method
df.write.copy_into_location(temp_stage, file_format_type="csv", partition_by="City")
Copy

  • options 매개 변수 수정:

이 경우 lineSep 및 dateFormat 은 RECORD_DELIMITER 및 DATE_FORMAT 으로 대체되며, 추가 권장 사항 섹션에는 모든 동등 항목이 있는 테이블이 있습니다.

다음은 RECORD_DELIMITER 및 DATE_FORMAT 으로 사전을 생성하고 해당 사전으로 options 메서드를 호출하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])
optionsParam = {"RECORD_DELIMITER": "\r\n", "DATE_FORMAT": "YYYY/MM/DD"}

## Using csv method
df.write.csv(stage, format_type_options=optionsParam)

## Using copy_into_location method
df.write.csv(stage, file_format_type="csv", format_type_options=optionsParam)
Copy

시나리오 2

입력 코드

다음은 JSON 형식으로 데이터를 저장하는 예제입니다.

path_json_file = "/path/to/file.json"

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

df.write.save(path_json_file, format="json")
df.write.save(path_json_file, format="json", mode="overwrite")
df.write.save(path_json_file, format="json", mode="overwrite", dateFormat="YYYY/MM/DD", timestampFormat="YYYY-MM-DD HH24:MI:SS.FF3")
df.write.save(path_json_file, format="json", mode="overwrite", partitionBy="City", dateFormat="YYYY/MM/DD", timestampFormat="YYYY-MM-DD HH24:MI:SS.FF3")
Copy

출력 코드

이 도구는 출력 코드에 EWI SPRKPY1083 을 추가하여 이 함수가 Snowpark에서 지원되지 않지만 우회 방법이 있음을 알려줍니다.

path_json_file = "/path/to/file.json"

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_json_file, format="json")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_json_file, format="json", mode="overwrite")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_json_file, format="json", mode="overwrite", dateFormat="YYYY/MM/DD", timestampFormat="YYYY-MM-DD HH24:MI:SS.FF3")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_json_file, format="json", mode="overwrite", partitionBy="City", dateFormat="YYYY/MM/DD", timestampFormat="YYYY-MM-DD HH24:MI:SS.FF3")
Copy

권장 수정

해결 방법으로 Snowpark DataFrameReader 메서드를 대신 사용할 수 있습니다.

  • path 및 format 매개 변수 수정:

    • load 메서드를 json 또는 copy_into_location 메서드로 바꿉니다

    • copy_into_location 메서드를 사용하는 경우 file_format_type 매개 변수 를 사용하여 형식을 지정해야 합니다.

    • 첫 번째 매개 변수 path 는 Snowpark 와 동등하게 생성하기 위해 스테이지에 있어야 합니다.

아래는 임시 스테이지를 생성하고 그 안에 파일을 넣은 다음 위에서 언급한 메서드 중 하나를 호출하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Stage creation

temp_stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
my_session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}').show()
my_session.file.put(f"file:///path/to/file.json", f"@{temp_stage}")
stage_file_path = f"{temp_stage}file.json"

## Using json method
df.write.json(stage_file_path)

## Using copy_into_location method
df.write.copy_into_location(stage_file_path, file_format_type="json")
Copy

아래는 데이지 체인에 mode 메서드를 추가하고 overwrite 를 매개 변수로 추가하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Using json method
df.write.mode("overwrite").json(temp_stage)

## Using copy_into_location method
df.write.mode("overwrite").copy_into_location(temp_stage, file_format_type="json")
Copy

  • partitionBy 매개 변수:

    • 다음과 같이 CSV 메서드에서 partition_by 매개 변수를 사용합니다.

다음은 CSV 메서드에서 partition_by 매개 변수를 사용한 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Using json method
df.write.json(temp_stage, partition_by="City")

## Using copy_into_location method
df.write.copy_into_location(temp_stage, file_format_type="json", partition_by="City")
Copy

  • options 매개 변수 수정:

이 경우 dateFormat 및 timestampFormat 은 DATE_FORMAT 및 TIMESTAMP_FORMAT 로 대체되며, 추가 권장 사항 섹션에는 모든 동등 매개 변수가 있는 테이블이 있습니다.

다음은 DATE_FORMAT 및 TIMESTAMP_FORMAT 으로 사전을 생성하고 해당 사전으로 options 메서드를 호출하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])
optionsParam = {"DATE_FORMAT": "YYYY/MM/DD", "TIMESTAMP_FORMAT": "YYYY-MM-DD HH24:MI:SS.FF3"}

## Using json method
df.write.json(stage, format_type_options=optionsParam)

## Using copy_into_location method
df.write.copy_into_location(stage, file_format_type="json", format_type_options=optionsParam)
Copy

시나리오 3

입력 코드

다음은 PARQUET 형식으로 데이터를 저장하는 예제입니다.

path_parquet_file = "/path/to/file.parquet"

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

df.write.save(path_parquet_file, format="parquet")
df.write.save(path_parquet_file, format="parquet", mode="overwrite")
df.write.save(path_parquet_file, format="parquet", mode="overwrite", pathGlobFilter="*.parquet")
df.write.save(path_parquet_file, format="parquet", mode="overwrite", partitionBy="City", pathGlobFilter="*.parquet")
Copy

출력 코드

이 도구는 출력 코드에 EWI SPRKPY1083 을 추가하여 이 함수가 Snowpark에서 지원되지 않지만 우회 방법이 있음을 알려줍니다.

path_parquet_file = "/path/to/file.parquet"

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_parquet_file, format="parquet")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_parquet_file, format="parquet", mode="overwrite")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_parquet_file, format="parquet", mode="overwrite", pathGlobFilter="*.parquet")
#EWI: SPRKPY1083 => The pyspark.sql.readwriter.DataFrameWriter.save function is not supported. A workaround is to use Snowpark DataFrameWriter copy_into_location method instead.
df.write.save(path_parquet_file, format="parquet", mode="overwrite", partitionBy="City", pathGlobFilter="*.parquet")
Copy

권장 수정

해결 방법으로 Snowpark DataFrameReader 메서드를 대신 사용할 수 있습니다.

  • path 및 format 매개 변수 수정:

    • load 메서드를 Parquet 또는 copy_into_location 메서드로 바꿉니다.

    • copy_into_location 메서드를 사용하는 경우 file_format_type 매개 변수 를 사용하여 형식을 지정해야 합니다.

    • 첫 번째 매개 변수 path 는 Snowpark 와 동등하게 생성하기 위해 스테이지에 있어야 합니다.

아래는 임시 스테이지를 생성하고 그 안에 파일을 넣은 다음 위에서 언급한 메서드 중 하나를 호출하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Stage creation

temp_stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
my_session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}').show()
my_session.file.put(f"file:///path/to/file.parquet", f"@{temp_stage}")
stage_file_path = f"{temp_stage}file.parquet"

## Using parquet method
df.write.parquet(stage_file_path)

## Using copy_into_location method
df.write.copy_into_location(stage, file_format_type="parquet")
Copy

아래는 데이지 체인에 mode 메서드를 추가하고 overwrite 를 매개 변수로 추가하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Using parquet method
df.write.mode("overwrite").parquet(temp_stage)

## Using copy_into_location method
df.write.mode("overwrite").copy_into_location(stage, file_format_type="parquet")
Copy

  • partitionBy 매개 변수:

    • 다음과 같이 CSV 메서드에서 partition_by 매개 변수를 사용합니다.

아래는 parquet 메서드에서 partition_by 매개 변수를 사용한 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

## Using parquet method
df.write.parquet(temp_stage, partition_by="City")

## Using copy_into_location method
df.write.copy_into_location(stage, file_format_type="parquet", partition_by="City")
Copy

  • options 매개 변수 수정:

이 경우 pathGlobFilter 는 PATTERN 으로 대체되며, 추가 권장 사항 섹션에는 모든 동긍 매개 변수가 포함된 테이블이 있습니다.

다음은 PATTERN 을 사용하여 사전을 생성하고 해당 사전으로 options 메서드를 호출하는 예제입니다.

data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]
df = spark.createDataFrame(data, schema=["Name", "Age", "City"])
optionsParam = {"PATTERN": "*.parquet"}

## Using parquet method
df.write.parquet(stage, format_type_options=optionsParam)

## Using copy_into_location method
df.write.copy_into_location(stage, file_format_type="parquet", format_type_options=optionsParam)
Copy

추가 권장 사항

  • Spark와 Snowpark의 옵션은 동일하지 않지만 매핑할 수 있습니다.

Spark 옵션

가능한 값

Snowpark 등가물

설명

헤더

true 또는 false

SKIP_HEADER = 1 / SKIP_HEADER = 0

파일의 첫 라인을 열 이름으로 사용하려면 다음과 같이 하십시오.

구분_기호

모든 단일/다중 문자 필드 구분 기호

FIELD_DELIMITER

각 열/필드에 단일/복수의 문자를 구분 기호로 지정합니다.

sep

단일 문자 필드 구분 기호

FIELD_DELIMITER

각 열/필드에 단일 문자를 구분 기호로 지정합니다.

인코딩

UTF-8, UTF-16 등…

ENCODING

지정된 인코딩 유형으로 CSV 파일을 디코딩합니다. 기본 인코딩은 UTF-8입니다

lineSep

단일 문자 라인 바꿈 기호

RECORD_DELIMITER

파일 구문 분석에 사용할 라인 바꿈 구분자를 정의합니다.

pathGlobFilter

파일 패턴

PATTERN

패턴과 일치하는 파일 이름만 읽도록 패턴을 정의합니다.

recursiveFileLookup

true 또는 false

N/A

디렉터리를 재귀적으로 스캔하여 파일을 읽으려면. 이 옵션의 기본값은 false입니다.

인용

작은따옴표로 묶을 1개 문자

FIELD_OPTIONALLY_ENCLOSED_BY

구분 기호/숫자가 값의 일부가 될 수 있는 필드가 포함된 필드/열을 인용하려면. QuoteAll 옵션과 함께 사용할 때 모든 필드를 인용하려면 이 문자를 입력합니다. 이 옵션의 기본값은 큰따옴표(“)입니다.

nullValue

Null을 대체할 문자열

NULL_IF

데이터프레임을 읽고 쓰는 동안 문자열을 null 값으로 바꿉니다.

날짜 형식

유효한 날짜 형식

DATE_FORMAT

날짜 형식을 나타내는 문자열을 정의합니다. 기본 형식은 yyyy-MM-dd입니다.

타임스탬프 형식

유효한 타임스탬프 형식

TIMESTAMP_FORMAT

타임스탬프 형식을 나타내는 문자열을 정의합니다. 기본 형식은 yyyy-MM-dd’T’HH:mm:ss입니다.

이스케이프

모든 단일 문자

ESCAPE

단일 문자를 이스케이프 문자(\)로 설정하여 기본 이스케이프 문자를 재정의하려면.

추론 스키마

true 또는 false

INFER_SCHEMA

파일 스키마 자동 감지

mergeSchema

true 또는 false

N/A

이는 infer_schema가 Parquet 파일 구조를 결정할 때마다 발생하므로 Snowflake에서는 필요하지 않습니다

  • modifiedBefore / modifiedAfter 옵션의 경우 메타데이터 열을 사용하여 Snowflake에서 동일한 결과를 얻은 다음 df.filter(METADATA_FILE_LAST_MODIFIED > ‘some_date’) 와 같은 필터를 추가할 수 있습니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1044

경고

이 문제 코드는 Spark Conversion Core Version 2.4.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.functions.split에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.split 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

메서드에 전달되는 매개 변수의 양에 따라 몇 가지 시나리오가 있습니다.

시나리오 1

입력

다음은 split 함수에 _str_및 _pattern_매개 변수만 있는 경우의 예입니다

F.split('col', '\\|')
Copy

출력

이 도구에는 해결 방법이 있음을 나타내는 EWI SPRKPY1044 메시지가 표시됩니다.

#EWI: SPRKPY1044 => pyspark.sql.functions.split has a workaround, see the documentation for more info
F.split('col', '\\|')
Copy

권장 수정

해결 방법으로 패턴 매개 변수를 사용하여 Snowflake.snowpark.functions.lit 함수를 호출하여 분할로 전송할 수 있습니다.

F.split('col', lit('\\|'))
## the result of lit will be sent to the split function
Copy

시나리오 2

입력

다음은 split 함수에 _str, 패턴, limit_매개 변수가 있는 또 다른 예입니다.

F.split('col', '\\|', 2)
Copy

출력

이 도구에는 해결 방법이 있음을 나타내는 EWI SPRKPY1044 메시지가 표시됩니다.

#EWI: SPRKPY1044 => pyspark.sql.functions.split has a workaround, see the documentation for more info
F.split('col', '\\|', 2)
Copy

권장 수정

이 특정 시나리오는 지원되지 않습니다.

추가 권장 사항

SPRKPY1015

메시지: pyspark.sql.functions.atanh에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.atanh 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제에서 pyspark는 pyspark.sql.functions.atanh 를 사용하여 데이터 프레임의 atanh를 계산합니다.

from pyspark.sql import SparkSession
from pyspark.sql.functions import atanh
spark = SparkSession.builder.getOrCreate()
data = [['V1', 0.14],
        ['V2', 0.32],
        ['V3', 0.4],
        ['V4', -0.36]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
df_result = df.withColumn("atanh_value", atanh(df["value"]))
Copy

출력

SMA 는 atanh이 사용되는 라인 위에 EWI SPRKPY1015 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

from snowflake.snowpark import Session

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['V1', 0.14],
        ['V2', 0.32],
        ['V3', 0.4],
        ['V4', -0.36]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
#EWI: SPRKPY1015 => pyspark.sql.functions.atanh has a workaround, see documentation for more info
df_result = df.withColumn("atanh_value", atanh(df["value"]))
Copy

권장 수정

직접적인 “atanh” 구현은 없지만 대신 “call_function”을 사용할 수 있으며, 첫 번째 매개 변수로 “atanh”를 사용하고 두 번째 매개 변수로 colName을 사용할 수 있습니다.

import snowflake.snowpark as snowpark
from snowflake.snowpark import Session
from snowflake.snowpark.functions import call_function, col

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['V1', 0.14],
        ['V2', 0.32],
        ['V3', 0.4],
        ['V4', -0.36]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
df_result = df.select(call_function('atanh', col('value')))
Copy

추가 권장 사항

SPRKPY1005

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

메시지 : pyspark.conf.SparkConf 는 필수가 아닙니다.

카테고리 : 경고.

설명

이 문제는 도구가 필수가 아닌 pyspark.conf.SparkConf 가 사용되는 것을 감지할 때 발생합니다.

시나리오

입력

SparkConf 는 매개 변수 없이 또는 loadDefaults와 함께 호출할 수 있습니다.

from pyspark import SparkConf

my_conf = SparkConf(loadDefaults=True)
Copy

출력

두 경우 모두 (매개 변수 유무 상관 없음) SMA 는 Snowpark Session.builder 오브젝트를 생성합니다.

#EWI: SPRKPY1005 => pyspark.conf.SparkConf is not required
#from pyspark import SparkConf
pass

#EWI: SPRKPY1005 => pyspark.conf.SparkConf is not required
my_conf = Session.builder.configs({"user" : "my_user", "password" : "my_password", "account" : "my_account", "role" : "my_role", "warehouse" : "my_warehouse", "database" : "my_database", "schema" : "my_schema"}).create()
Copy

추가 권장 사항

  • 불필요한 매개 변수가 제거되고 경고 설명이 삽입됩니다. 사용자의 추가 작업은 없어야 합니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1054

메시지: pyspark.sql.readwriter.DataFrameReader.format은 지원되지 않습니다.

카테고리: 경고.

설명

이 문제는 pyspark.sql.readwriter.DataFrameReader.format 에 Snowpark에서 지원하지 않는 인자가 있을 때 표시됩니다.

시나리오

로딩하려는 형식의 유형에 따라 몇 가지 시나리오가 있습니다. 이는 지원되는 또는 지원되지 않는 형식일 수 있습니다.

시나리오 1

입력

이 도구는 로딩하려는 형식의 유형을 분석하며, 지원되는 형식은 다음과 같습니다.

  • Csv

  • JSON

  • Parquet

  • Orc

아래 예는 Csv 값을 전달할 때 도구가 format 메서드를 변환하는 방법을 보여줍니다.

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

df1 = spark.read.format('csv').load('/path/to/file')
Copy

출력

이 도구는 format 메서드를 Csv 메서드 호출로 변환합니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()

df1 = spark.read.csv('/path/to/file')
Copy

권장 수정

이 경우 도구에 EWI 가 표시되지 않으므로 수정할 필요가 없습니다.

시나리오 2

입력

아래 예는 Jdbc 값을 전달할 때 도구가 format 메서드를 변환하는 방법을 보여줍니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()

df2 = spark.read.format('jdbc') \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("url", "jdbc:mysql://localhost:3306/emp") \
    .option("dbtable", "employee") \
    .option("user", "root") \
    .option("password", "root") \
    .load()
Copy

출력

이 도구는 “jdbc” 값이 지원되지 않음을 나타내는 EWI SPRKPY1054 를 표시합니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()

#EWI: SPRKPY1054 => pyspark.sql.readwriter.DataFrameReader.format with argument value "jdbc" is not supported.
#EWI: SPRKPY1002 => pyspark.sql.readwriter.DataFrameReader.load is not supported

df2 = spark.read.format('jdbc') \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("url", "jdbc:mysql://localhost:3306/emp") \
    .option("dbtable", "employee") \
    .option("user", "root") \
    .option("password", "root") \
    .load()
Copy

권장 수정

지원되지 않는 시나리오의 경우 읽을 파일에 따라 종속성이 달라지므로 특별한 수정 사항이 없습니다.

시나리오 3

입력

아래 예제는 CSV 를 전달할 때 이 도구가 format 메서드를 변수를 사용하여 변환하는 방법을 보여줍니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()

myFormat = 'csv'
df3 = spark.read.format(myFormat).load('/path/to/file')
Copy

출력

도구가 런타임에 변수 값을 확인할 수 없으므로 해당 값 “”이 지원되지 않음을 나타내는 EWI SPRKPY1054 를 표시합니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()

myFormat = 'csv'
#EWI: SPRKPY1054 => pyspark.sql.readwriter.DataFrameReader.format with argument value "" is not supported.
#EWI: SPRKPY1002 => pyspark.sql.readwriter.DataFrameReader.load is not supported
df3 = spark.read.format(myFormat).load('/path/to/file')
Copy

권장 수정

해결 방법으로 변수 값을 확인하여 format 호출에 문자열로 추가할 수 있습니다.

추가 권장 사항

  • Snowpark 위치는 Snowflake 스테이지 를 사용하는 클라우드 위치만 허용합니다.

  • Snowpark가 지원하는 메서드에 대한 설명서는 설명서 에서 확인할 수 있습니다

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1060

메시지: 인증 메커니즘은 connection.json(템플릿 제공)입니다.

카테고리: 경고.

설명

이 문제는 도구가 pyspark.conf.SparkConf 의 사용을 감지할 때 발생합니다.

시나리오

입력

Snowpark에서는 인증 메커니즘이 다르므로 이 도구는 해당 사용처를 제거하고 대신 연결 구성 파일(connection.json) 을 생성합니다.

from pyspark import SparkConf

my_conf = SparkConf(loadDefaults=True)
Copy

출력

이 도구는 인증 메커니즘이 다르다는 것을 나타내는 EWI SPRKPY1060 을 추가합니다.

#EWI: SPRKPY1002 => pyspark.conf.SparkConf is not supported
#EWI: SPRKPY1060 => The authentication mechanism is connection.json (template provided).
#my_conf = Session.builder.configs(connection_parameter).getOrCreate()

my_conf = None
Copy

권장 수정

연결을 생성하려면 connection.json 파일에 정보를 입력해야 합니다.

{
  "user": "<USER>",
  "password": "<PASSWORD>",
  "account": "<ACCOUNT>",
  "role": "<ROLE>",
  "warehouse": "<WAREHOUSE>",
  "database": "<DATABASE>",
  "schema": "<SCHEMA>"
}
Copy

추가 권장 사항

SPRKPY1031

경고

이 문제 코드는 Spark Conversion Core 2.7.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.column.Column.contains에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.column.Column.contains 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.column.Column.contains 함수의 사용 예제입니다. 이 예제에서는 contains 함수를 사용하여 ‘City’ 열에 하위 문자열 ‘New’가 포함된 행을 필터링합니다.

df = spark.createDataFrame([("Alice", "New York"), ("Bob", "Los Angeles"), ("Charlie", "Chicago")], ["Name", "City"])
df_filtered = df.filter(col("City").contains("New"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1031 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([("Alice", "New York"), ("Bob", "Los Angeles"), ("Charlie", "Chicago")], ["Name", "City"])
#EWI: SPRKPY1031 => pyspark.sql.column.Column.contains has a workaround, see documentation for more info
df_filtered = df.filter(col("City").contains("New"))
Copy

권장 수정

해결 방법으로 열을 첫 번째 인자로, 검색할 요소를 두 번째 인자로 전달하여 Snowflake.snowpark.functions.contains 함수를 사용할 수 있습니다. 검색할 요소가 리터럴 값인 경우 lit 함수를 사용하여 열 식으로 변환해야 합니다.

from snowflake.snowpark import functions as f
df = spark.createDataFrame([("Alice", "New York"), ("Bob", "Los Angeles"), ("Charlie", "Chicago")], ["Name", "City"])
df_filtered = df.filter(f.contains(col("City"), f.lit("New")))
Copy

추가 권장 사항

SPRKPY1020

메시지: pyspark.sql.functions.instr에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.instr 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

다음은 pyspark 명령의 기본적인 사용 예시입니다.

from pyspark.sql import SparkSession
from pyspark.sql.functions import instr
spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame([('abcd',)], ['test',])
df.select(instr(df.test, 'cd').alias('result')).collect()
Copy

출력:

SMA 는 instr이 사용되는 라인 위에 EWI SPRKPY1020 을 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

from snowflake.snowpark import Session

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
df = spark.createDataFrame([('abcd',)], ['test',])
#EWI: SPRKPY1020 => pyspark.sql.functions.instr has a workaround, see documentation for more info
df.select(instr(df.test, 'cd').alias('result')).collect()
Copy

권장 수정

함수 charindex 를 사용하고 처음 두 매개 변수의 순서를 변경하여 수동으로 변경해야 합니다.

import snowflake.snowpark as snowpark
from snowflake.snowpark import Session
from snowflake.snowpark.functions import charindex, lit

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
df = spark.createDataFrame([('abcd',)], ['test',])
df.select(charindex(lit('cd'), df.test).as_('result')).show()
Copy

추가 권장 사항

SPRKPY1071

메시지: pyspark.rdd.RDD.getNumPartitions 함수는 Snowpark에서 필수가 아닙니다. 따라서 모든 참조를 제거해야 합니다.

카테고리: 경고

설명

이 문제는 도구가 pyspark.rdd.RDD.getNumPartitions 함수의 사용을 감지할 때 발생합니다. Snowflake는 마이크로 파티션 메커니즘을 사용하므로 이 함수를 사용할 필요가 없습니다.

시나리오

입력

getNumPartitions 은 RDD 에서 파티션의 수량을 반환합니다.

df = spark.createDataFrame([('2015-04-08',), ('5',), [Row(a=1, b="b")]], ['dt', 'num', 'row'])

print(df.getNumPartitions())
Copy

출력

이 도구는 EWI 를 추가하여 getNumPartitions 이 필수가 아님을 알려줍니다.

df = spark.createDataFrame([('2015-04-08',), ('5',), [Row(a=1, b="b")]], ['dt', 'num', 'row'])
#EWI: SPRKPY1071 => The getNumPartitions are not required in Snowpark. So, you should remove all references.

print(df.getNumPartitions())
Copy

권장 수정

이 함수의 모든 사용을 제거합니다.

df = spark.createDataFrame([('2015-04-08',), ('5',), [Row(a=1, b="b")]], ['dt', 'num', 'row'])
Copy

추가 권장 사항

SPRKPY1082

메시지: DataFrameReader.load 함수는 지원되지 않습니다. 해결 방법은 대신 Snowpark DataFrameReader 형식별 메서드(avro csv, json, orc, parquet)를 사용하는 것입니다. 경로 매개 변수는 스테이지 위치여야 합니다.

카테고리: 경고

설명

DataFrameReader.load 함수는 지원되지 않습니다. 해결 방법은 대신 Snowpark DataFrameReader 메서드를 사용하는 것입니다.

시나리오

이 메서드 DataFrameReader.load(path, format, schema, **옵션) 에 대한 Spark 서명은 Snowpark에 존재하지 않습니다. 따라서 로딩 함수를 사용하는 모든 출력 코드에는 EWI 가 포함됩니다.

시나리오 1

입력

다음은 CSV 소스에서 데이터 로딩을 시도하는 예제입니다.

path_csv_file = "/path/to/file.csv"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])

my_session.read.load(path_csv_file, "csv").show()
my_session.read.load(path_csv_file, "csv", schema=schemaParam).show()
my_session.read.load(path_csv_file, "csv", schema=schemaParam, lineSep="\r\n", dateFormat="YYYY/MM/DD").show()
Copy

출력

SMA 에는 EWI SPRKPY1082 를 추가하여 이 함수가 Snowpark에서 지원되지 않지만 해결 방법이 있음을 알려줍니다.

path_csv_file = "/path/to/file.csv"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.

my_session.read.load(path_csv_file, "csv").show()
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.
my_session.read.load(path_csv_file, "csv", schema=schemaParam).show()
#EWI: The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.
my_session.read.load(path_csv_file, "csv", schema=schemaParam, lineSep="\r\n", dateFormat="YYYY/MM/DD").show()
Copy

권장 수정

해결 방법으로 Snowpark DataFrameReader 메서드를 대신 사용할 수 있습니다.

  • path 및 format 매개 변수 수정:

    • load 메서드를 csv 메서드로 바꿉니다.

    • 첫 번째 매개 변수 path 는 Snowpark 와 동등하게 생성하기 위해 스테이지에 있어야 합니다.

다음은 임시 스테이지를 생성하고 여기에 파일을 넣은 다음 CSV 메서드를 호출하는 예제입니다.

path_csv_file = "/path/to/file.csv"

## Stage creation

temp_stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
my_session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}').show()
my_session.file.put(f"file:///path/to/file.csv", f"@{temp_stage}")
stage_file_path = f"{temp_stage}file.csv"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])

my_session.read.csv(stage_file_path).show()
Copy

  • schema 매개 변수 수정:

    • 스키마는 다음과 같이 schema 함수를 사용하여 설정할 수 있습니다.

schemaParam = StructType([
        StructField("name", StringType(), True),
        StructField("city", StringType(), True)
    ])

df = my_session.read.schema(schemaParam).csv(temp_stage)
Copy

  • options 매개 변수 수정:

spark와 snowpark 간의 옵션 은 동일하지 않으며, 이 경우 lineSep 및 dateFormat 은 RECORD_DELIMITER 및 DATE_FORMAT 으로 대체되며, 추가 권장 사항 섹션에는 모든 동등한 옵션이 있는 테이블이 있습니다.

다음은 RECORD_DELIMITER 및 DATE_FORMAT 으로 사전을 생성하고 해당 사전으로 options 메서드를 호출하는 예제입니다.

optionsParam = {"RECORD_DELIMITER": "\r\n", "DATE_FORMAT": "YYYY/MM/DD"}
df = my_session.read.options(optionsParam).csv(stage)
Copy

시나리오 2

입력

다음은 JSON 소스에서 데이터 로딩을 시도하는 예제입니다.

path_json_file = "/path/to/file.json"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])

my_session.read.load(path_json_file, "json").show()
my_session.read.load(path_json_file, "json", schema=schemaParam).show()
my_session.read.load(path_json_file, "json", schema=schemaParam, dateFormat="YYYY/MM/DD", timestampFormat="YYYY-MM-DD HH24:MI:SS.FF3").show()
Copy

출력

SMA 에는 EWI SPRKPY1082 를 추가하여 이 함수가 Snowpark에서 지원되지 않지만 해결 방법이 있음을 알려줍니다.

path_json_file = "/path/to/file.json"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.

my_session.read.load(path_json_file, "json").show()
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.
my_session.read.load(path_json_file, "json", schema=schemaParam).show()
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.
my_session.read.load(path_json_file, "json", schema=schemaParam, dateFormat="YYYY/MM/DD", timestampFormat="YYYY-MM-DD HH24:MI:SS.FF3").show()
Copy

권장 수정

해결 방법으로 Snowpark DataFrameReader 메서드를 대신 사용할 수 있습니다.

  • path 및 format 매개 변수 수정:

    • load 메서드를 json 메서드로 바꿉니다

    • 첫 번째 매개 변수 path 는 Snowpark 와 동등하게 생성하기 위해 스테이지에 있어야 합니다.

다음은 임시 스테이지를 생성하고 여기에 파일을 넣은 다음 JSON 메서드를 호출하는 예제입니다.

path_json_file = "/path/to/file.json"

## Stage creation

temp_stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
my_session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}').show()
my_session.file.put(f"file:///path/to/file.json", f"@{temp_stage}")
stage_file_path = f"{temp_stage}file.json"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])

my_session.read.json(stage_file_path).show()
Copy

  • schema 매개 변수 수정:

    • 스키마는 다음과 같이 schema 함수를 사용하여 설정할 수 있습니다.

schemaParam = StructType([
        StructField("name", StringType(), True),
        StructField("city", StringType(), True)
    ])

df = my_session.read.schema(schemaParam).json(temp_stage)
Copy

  • options 매개 변수 수정:

Spark와 snowpark 간의 옵션 은 동일하지 않으며, 이 경우 dateFormat 및 timestampFormat 은 DATE_FORMAT 및 TIMESTAMP_FORMAT 으로 대체되며, 추가 권장 사항 섹션에는 모든 동등한 옵션이 있는 테이블이 있습니다.

다음은 DATE_FORMAT 및 TIMESTAMP_FORMAT 으로 사전을 생성하고 해당 사전으로 options 메서드를 호출하는 예제입니다.

optionsParam = {"DATE_FORMAT": "YYYY/MM/DD", "TIMESTAMP_FORMAT": "YYYY-MM-DD HH24:MI:SS.FF3"}
df = Session.read.options(optionsParam).json(stage)
Copy

시나리오 3

입력

다음은 PARQUET 소스에서 데이터 로딩을 시도하는 예제입니다.

path_parquet_file = "/path/to/file.parquet"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])

my_session.read.load(path_parquet_file, "parquet").show()
my_session.read.load(path_parquet_file, "parquet", schema=schemaParam).show()
my_session.read.load(path_parquet_file, "parquet", schema=schemaParam, pathGlobFilter="*.parquet").show()
Copy

출력

SMA 에는 EWI SPRKPY1082 를 추가하여 이 함수가 Snowpark에서 지원되지 않지만 해결 방법이 있음을 알려줍니다.

path_parquet_file = "/path/to/file.parquet"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.

my_session.read.load(path_parquet_file, "parquet").show()
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.
my_session.read.load(path_parquet_file, "parquet", schema=schemaParam).show()
#EWI: SPRKPY1082 => The pyspark.sql.readwriter.DataFrameReader.load function is not supported. A workaround is to use Snowpark DataFrameReader format specific method instead (avro csv, json, orc, parquet). The path parameter should be a stage location.
my_session.read.load(path_parquet_file, "parquet", schema=schemaParam, pathGlobFilter="*.parquet").show()
Copy

권장 수정

해결 방법으로 Snowpark DataFrameReader 메서드를 대신 사용할 수 있습니다.

  • path 및 format 매개 변수 수정:

    • load 메서드를 parquet 메서드로 바꿉니다

    • 첫 번째 매개 변수 path 는 Snowpark 와 동등하게 생성하기 위해 스테이지에 있어야 합니다.

다음은 임시 스테이지를 생성하고 여기에 파일을 넣은 다음 PARQUET 메서드를 호출하는 예제입니다.

path_parquet_file = "/path/to/file.parquet"

## Stage creation

temp_stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
my_session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}').show()
my_session.file.put(f"file:///path/to/file.parquet", f"@{temp_stage}")
stage_file_path = f"{temp_stage}file.parquet"

schemaParam = StructType([
        StructField("Name", StringType(), True),
        StructField("Superhero", StringType(), True)
    ])

my_session.read.parquet(stage_file_path).show()
Copy

  • schema 매개 변수 수정:

    • 스키마는 다음과 같이 schema 함수를 사용하여 설정할 수 있습니다.

schemaParam = StructType([
        StructField("name", StringType(), True),
        StructField("city", StringType(), True)
    ])

df = my_session.read.schema(schemaParam).parquet(temp_stage)
Copy

  • options 매개 변수 수정:

Spark와 snowpark 간의 옵션 은 동일하지 않으며, 이 경우 pathGlobFilter 는 PATTERN 으로 대체되고, 추가 권장 사항 섹션에는 모든 동등한 옵션이 있는 테이블이 있습니다.

다음은 PATTERN 을 사용하여 사전을 생성하고 해당 사전으로 options 메서드를 호출하는 예제입니다.

optionsParam = {"PATTERN": "*.parquet"}
df = Session.read.options(optionsParam).parquet(stage)
Copy

추가 권장 사항

  • Spark와 Snowpark의 옵션은 동일하지 않지만 매핑이 가능하다는 점을 고려하십시오.

Spark 옵션

가능한 값

Snowpark 등가물

설명

헤더

true 또는 false

SKIP_HEADER = 1 / SKIP_HEADER = 0

파일의 첫 라인을 열 이름으로 사용하려면 다음과 같이 하십시오.

구분_기호

모든 단일/다중 문자 필드 구분 기호

FIELD_DELIMITER

각 열/필드에 단일/복수의 문자를 구분 기호로 지정합니다.

sep

단일 문자 필드 구분 기호

FIELD_DELIMITER

각 열/필드에 단일 문자를 구분 기호로 지정합니다.

인코딩

UTF-8, UTF-16 등…

ENCODING

지정된 인코딩 유형으로 CSV 파일을 디코딩합니다. 기본 인코딩은 UTF-8입니다

lineSep

단일 문자 라인 바꿈 기호

RECORD_DELIMITER

파일 구문 분석에 사용할 라인 바꿈 구분자를 정의합니다.

pathGlobFilter

파일 패턴

PATTERN

패턴과 일치하는 파일 이름만 읽도록 패턴을 정의합니다.

recursiveFileLookup

true 또는 false

N/A

디렉터리를 재귀적으로 스캔하여 파일을 읽으려면. 이 옵션의 기본값은 false입니다.

인용

작은따옴표로 묶을 1개 문자

FIELD_OPTIONALLY_ENCLOSED_BY

구분 기호/숫자가 값의 일부가 될 수 있는 필드가 포함된 필드/열을 인용하려면. QuoteAll 옵션과 함께 사용할 때 모든 필드를 인용하려면 이 문자를 입력합니다. 이 옵션의 기본값은 큰따옴표(“)입니다.

nullValue

Null을 대체할 문자열

NULL_IF

데이터프레임을 읽고 쓰는 동안 문자열을 null 값으로 바꿉니다.

날짜 형식

유효한 날짜 형식

DATE_FORMAT

날짜 형식을 나타내는 문자열을 정의합니다. 기본 형식은 yyyy-MM-dd입니다.

타임스탬프 형식

유효한 타임스탬프 형식

TIMESTAMP_FORMAT

타임스탬프 형식을 나타내는 문자열을 정의합니다. 기본 형식은 yyyy-MM-dd’T’HH:mm:ss입니다.

이스케이프

모든 단일 문자

ESCAPE

단일 문자를 이스케이프 문자(\)로 설정하여 기본 이스케이프 문자를 재정의하려면.

추론 스키마

true 또는 false

INFER_SCHEMA

파일 스키마 자동 감지

mergeSchema

true 또는 false

N/A

이는 infer_schema가 Parquet 파일 구조를 결정할 때마다 발생하므로 Snowflake에서는 필요하지 않습니다

  • modifiedBefore / modifiedAfter 옵션의 경우 메타데이터 열을 사용하여 Snowflake에서 동일한 결과를 얻은 다음 df.filter(METADATA_FILE_LAST_MODIFIED > ‘some_date’) 와 같은 필터를 추가할 수 있습니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1045

메시지: pyspark.sql.functions.map_values에 해결 방법이 있습니다

카테고리: 경고.

설명

이 함수는 map/dictionary (keys/values) 이 포함된 열에서 값 목록을 추출하는 데 사용됩니다.

이 문제는 도구가 pyspark.sql.functions.map_values 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

다음은 map_values 메서드 사용 예시입니다.

df = spark.createDataFrame(
    [(1, {'Apple': 'Fruit', 'Potato': 'Vegetable'})],
    ("id", "a_map"))

df.select(map_values("a_map")).show()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1045 을 추가합니다.

df = spark.createDataFrame(
    [(1, {'Apple': 'Fruit', 'Potato': 'Vegetable'})],
    ("id", "a_map"))
#EWI: SPRKPY1045 => pyspark.sql.functions.map_values has a workaround, see documentation for more info

df.select(map_values("a_map")).show()
Copy

권장 수정

해결 방법으로 udf 를 생성해 열의 값을 가져올 수 있습니다. 아래 예는 udf를 생성한 다음 F.map_values 에 할당하여 사용하는 방법을 보여줍니다.

from snowflake.snowpark import functions as F
from snowflake.snowpark.types import ArrayType, MapType

map_values_udf=None

def map_values(map):
    global map_values_udf
    if not map_values_udf:
        def _map_values(map: dict)->list:
            return list(map.values())
        map_values_udf = F.udf(_map_values,return_type=ArrayType(),input_types=[MapType()],name="map_values",is_permanent=False,replace=True)
    return map_values_udf(map)

F.map_values = map_values

df.select(map_values(colDict))
Copy

추가 권장 사항

SPRKPY1014

메시지: pyspark.sql.functions.asinh에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.asinh 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제에서 pyspark는 pyspark.sql.functions.asinh 를 사용하여 데이터 프레임의 asinh를 계산합니다.

from pyspark.sql import SparkSession
from pyspark.sql.functions import asinh
spark = SparkSession.builder.getOrCreate()
data = [['V1', 3.0],
        ['V2', 60.0],
        ['V3', 14.0],
        ['V4', 3.1]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
df_result = df.withColumn("asinh_value", asinh(df["value"]))
Copy

출력

SMA 는 asinh이 사용되는 라인 위에 EWI SPRKPY1014 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

from snowflake.snowpark import Session

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['V1', 3.0],
        ['V2', 60.0],
        ['V3', 14.0],
        ['V4', 3.1]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
#EWI: SPRKPY1014 => pyspark.sql.functions.asinh has a workaround, see documentation for more info
df_result = df.withColumn("asinh_value", asinh(df["value"]))
Copy

권장 수정

직접적인 “asinh” 구현은 없지만 대신 “call_function”을 사용할 수 있으며, 첫 번째 매개 변수로 “asinh”를 사용하고 두 번째 매개 변수로 colName을 사용할 수 있습니다.

import snowflake.snowpark as snowpark
from snowflake.snowpark import Session
from snowflake.snowpark.functions import call_function, col

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['V1', 3.0],
        ['V2', 60.0],
        ['V3', 14.0],
        ['V4', 3.1]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
df_result = df.select(call_function('asinh', col('value')))
Copy

추가 권장 사항

SPRKPY1004

메시지 : 기호 테이블을 로딩할 수 없습니다.

카테고리 : 구문 분석 오류입니다.

설명

이 문제는 도구 실행 과정에서 예기치 않은 오류가 발생했을 때 표시됩니다. 기호 테이블을 로딩할 수 없으므로 도구에서 평가 또는 변환 프로세스를 시작할 수 없습니다.

추가 권장 사항

  • 이는 소스 코드 자체의 오류라기보다는 도구가 소스 코드를 처리하는 방법의 오류일 가능성이 높습니다. SMA 지원팀으로 문의하는 것이 가장 좋은 해결책이 될 수 있습니다. sma-support@Snowflake.com으로 이메일을 보내주십시오.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 으로 이메일을 보내거나 SMA 에 문제를 게시할 수 있습니다.

SPRKPY1055

메시지: pyspark.sql.readwriter.DataFrameReader.option 키 값은 지원되지 않습니다.

카테고리: 경고.

설명

이 문제는 pyspark.sql.readwriter.DataFrameReader.option 키 값이 SnowFlake 에서 지원되지 않을 때 발생합니다.

이 도구는 옵션 호출 매개 변수를 분석하고 메서드(CSV 또는 JSON 또는 PARQUET)에 따라 키 값이 Snowpark에 해당하는 값이 있거나 없을 수 있으며, 모든 매개 변수에 해당하는 값이 있으면 도구는 EWI 를 추가하지 않고 해당 키 값으로 대체하고 그렇지 않으면 도구는 EWI 를 추가합니다.

동등 매개 기능 목록:

  • CSV 동등 기능:

Spark 옵션 키

Snowpark 동등 기능

sep

FIELD_DELIMITER

헤더

PARSE_HEADER

lineSep

RECORD_DELIMITER

pathGlobFilter

PATTERN

인용

FIELD_OPTIONALLY_ENCLOSED_BY

nullValue

NULL_IF

날짜 형식

DATE_FORMAT

타임스탬프 형식

TIMESTAMP_FORMAT

추론 스키마

INFER_SCHEMA

구분_기호

FIELD_DELIMITER

  • JSON 동등 기능:

Spark 옵션 키

Snowpark 동등 기능

날짜 형식

DATE_FORMAT

타임스탬프 형식

TIMESTAMP_FORMAT

pathGlobFilter

PATTERN

  • PARQUET 동등 기능:

Spark 옵션 키

Snowpark 동등 기능

pathGlobFilter

PATTERN

위 테이블에 없는 다른 키 옵션은 지원되지 않거나 Snowpark에 해당 옵션이 없습니다. 이 경우 도구는 매개 변수 정보와 함께 EWI 를 추가하고 체인에서 제거합니다.

시나리오

아래 시나리오는 CSV, JSON, PARQUET 에 적용됩니다.

option 메서드에 사용되는 키의 값에 따라 몇 가지 시나리오가 있습니다.

시나리오 1

입력

다음은 동등 키 를 사용하는 option 호출의 예입니다.

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

## CSV example:
spark.read.option("header", True).csv(csv_file_path)

## Json example:
spark.read.option("dateFormat", "dd-MM-yyyy").json(json_file_path)

## Parquet example:
spark.read.option("pathGlobFilter", "*.parquet").parquet(parquet_file_path)
Copy

출력

이 도구는 키를 올바른 등가물로 변환합니다.

from snowflake.snowpark import Session

spark = Session.builder.getOrCreate()

## CSV example:
spark.read.option("PARSE_HEADER", True).csv(csv_file_path)

## Json example:
spark.read.option("DATE_FORMAT", "dd-MM-yyyy").json(json_file_path)

## Parquet example:
spark.read.option("PATTERN", "*.parquet").parquet(parquet_file_path)
Copy

권장 수정

이 도구는 키의 값을 변환하므로 수정할 필요가 없습니다.

시나리오 2

입력

다음은 비등가 키 를 사용하는 option 호출의 예입니다.

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

## CSV example:
spark.read.option("anotherKeyValue", "myVal").csv(csv_file_path)

## Json example:
spark.read.option("anotherKeyValue", "myVal").json(json_file_path)

## Parquet example:
spark.read.option("anotherKeyValue", "myVal").parquet(parquet_file_path)
Copy

출력

이 도구는 해당 키가 지원되지 않음을 나타내는 EWI SPRKPY1055 를 추가하고 option 호출을 제거합니다.

from snowflake.snowpark import Session

spark = Session.builder.getOrCreate()

## CSV example:
#EWI: SPRKPY1055 => pyspark.sql.readwriter.DataFrameReader.option with key value "anotherKeyValue" is not supported.
spark.read.csv(csv_file_path)

## Json example:
#EWI: SPRKPY1055 => pyspark.sql.readwriter.DataFrameReader.option with key value "anotherKeyValue" is not supported.
spark.read.json(json_file_path)

## Parquet example:
#EWI: SPRKPY1055 => pyspark.sql.readwriter.DataFrameReader.option with key value "anotherKeyValue" is not supported.
spark.read.parquet(parquet_file_path)
Copy

권장 수정

변환 후 동작을 확인하는 것이 좋습니다.

추가 권장 사항

  • 동등하지 않은 매개 변수가 있는 경우 변환 후 동작을 확인하는 것이 좋습니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1061

메시지: Snowpark는 unix_timestamp 함수를 지원하지 않습니다

카테고리: 경고

설명

Snowpark에서 첫 번째 매개 변수는 필수이며, 도구가 매개 변수 없이 pyspark.sql.functions.unix_timestamp 의 사용을 감지하면 문제가 표시됩니다.

시나리오

입력

아래는 매개 변수 없이 unix_timestamp 메서드를 호출하는 예제입니다.

data = [["2015-04-08", "10"],["2015-04-10", "15"]]

df = spark.createDataFrame(data, ['dt', 'val'])
df.select(unix_timestamp()).show()
Copy

출력

이 함수 unix_timestamp(e: ColumnOrName, fmt: Optional["Column"] = None) 에 대한 Snowpark 서명으로, 첫 번째 매개 변수에서 알 수 있듯이 필수입니다.

이 도구는 EWI SPRKPY1061 을 추가하여 매개 변수가 없는 함수 unix_timestamp가 Snowpark에서 지원되지 않음을 알려줍니다.

data = [["2015-04-08", "10"],["2015-04-10", "15"]]

df = spark.createDataFrame(data, ['dt', 'val'])
#EWI: SPRKPY1061 => Snowpark does not support unix_timestamp functions with no parameters. See documentation for more info.
df.select(unix_timestamp()).show()
Copy

권장 수정

해결 방법으로 최소한 타임스탬프 문자열의 이름이나 열을 추가할 수 있습니다.

data = [["2015-04-08", "10"],["2015-04-10", "15"]]

df = spark.createDataFrame(data, ["dt", "val"])
df.select(unix_timestamp("dt")).show()
Copy

추가 권장 사항

SPRKPY1030

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.세션. SparkSession .Builder.appName에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.session.SparkSession.Builder.appName 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 이 EWI 를 생성하는 pyspark.sql.session.SparkSession.Builder.appName 함수의 사용 예제입니다. 이 예제에서는 appName 함수를 사용하여 MyApp 을 애플리케이션 이름으로 설정합니다.

session = SparkSession.builder.appName("MyApp").getOrCreate()
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1030 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

#EWI: SPRKPY1030 => pyspark.sql.session.SparkSession.Builder.appName has a workaround, see documentation for more info
session = Session.builder.appName("MyApp").getOrCreate()
Copy

권장 수정

해결 방법으로 appName 함수에 대한 확장자를 제공하는 snowpark_extensions 패키지를 가져오면 됩니다.

import snowpark_extensions
session = SessionBuilder.appName("MyApp").getOrCreate()
Copy

추가 권장 사항

SPRKPY1010

메시지: pyspark.sql.dataframe.DataFrame.checkpoint에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.dataframe.DataFrame.checkpoint 의 사용을 감지했을 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

PySpark 에서 체크포인트는 데이터프레임의 논리적 계획을 잘라내는 데 사용되며, 이는 논리적 계획이 커지는 것을 방지하기 위한 것입니다.

import tempfile
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
data = [['Q1', 300000],
        ['Q2', 60000],
        ['Q3', 500002],
        ['Q4', 130000]]

columns = ['Quarter', 'Score']
df = spark.createDataFrame(data, columns)
with tempfile.TemporaryDirectory() as d:
    spark.sparkContext.setCheckpointDir("/tmp/bb")
    df.checkpoint(False)
Copy

출력

SMA 는 접근 사분위수가 사용되는 라인 위에 EWI SPRKPY1010 을 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다. 또한 setCheckpointDir 은 지원되지 않는 것으로 표시되지만, 수정을 위해 체크포인트된 디렉터리는 필요하지 않습니다.

import tempfile
from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 300000],
        ['Q2', 60000],
        ['Q3', 500002],
        ['Q4', 130000]]

columns = ['Quarter', 'Score']
df = spark.createDataFrame(data, columns)
with tempfile.TemporaryDirectory() as d:
    #EWI: SPRKPY1002 => pyspark.context.SparkContext.setCheckpointDir is not supported
    spark.setCheckpointDir("/tmp/bb")
    #EWI: SPRKPY1010 => pyspark.sql.dataframe.DataFrame.checkpoint has a workaround, see documentation for more info
    df.checkpoint(False)
Copy

권장 수정

Snowpark를 사용하면 명시적 체크포인트가 필요하지 않습니다. Snowpark는 SQL 기반 작업과 함께 작동하므로 Snowflake 쿼리 최적화 엔진에 의해 최적화되어 불필요한 계산이나 제어할 수 없는 논리적 계획이 필요하지 않습니다.

그러나 데이터 프레임에서 계산 결과를 유지해야 하는 시나리오가 있을 수 있습니다. 이 시나리오에서는 Snowflake 테이블 또는 Snowflake 임시 테이블 에 데이터 프레임을 작성하여 결과를 구체화하여 저장할 수 있습니다.

  • 영구 테이블을 사용하거나 세션 종료 후에도 언제든지 계산된 결과에 액세스할 수 있습니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 300000],
        ['Q2', 60000],
        ['Q3', 500002],
        ['Q4', 130000]]

columns = ['Quarter', 'Score']
df = spark.createDataFrame(data, columns)
df.write.save_as_table("my_table", table_type="temporary") # Save the dataframe into Snowflake table "my_table".
df2 = Session.table("my_table") # Now I can access the stored result quering the table "my_table"
Copy

  • 다른 수정 방법인 임시 테이블을 사용하면 세션이 종료된 후 테이블이 삭제된다는 장점이 있습니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 300000],
        ['Q2', 60000],
        ['Q3', 500002],
        ['Q4', 130000]]

columns = ['Quarter', 'Score']
df = spark.createDataFrame(data, columns)
df.write.save_as_table("my_temp_table", table_type="temporary") # Save the dataframe into Snowflake table "my_temp_table".
df2 = Session.table("my_temp_table") # Now I can access the stored result quering the table "my_temp_table"
Copy

추가 권장 사항

SPRKPY1101

카테고리

구문 분석 오류가 발생했습니다.

설명

도구가 구문 분석 오류를 인식하면 이를 복구하려고 시도하고 다음 라인에서 프로세스를 계속 진행합니다. 이러한 경우 해당 라인에 오류와 설명이 표시됩니다.

이 예는 공백과 탭 간의 불일치 오류를 처리하는 방법을 보여줍니다.

입력 코드

def foo():
    x = 5 # Spaces
	y = 6 # Tab

def foo2():
    x=6
    y=7
Copy

출력 코드

def foo():
    x = 5 # Spaces
## EWI: SPRKPY1101 => Unrecognized or invalid CODE STATEMENT @(3, 2). Last valid token was '5' @(2, 9), failed token 'y' @(3, 2)
## y = 6 # Tab

def foo2():
    x=6
    y=7
Copy

권장 사항

  • 설명이 달린 라인을 수정해 보십시오.

  • 자세한 지원이 필요하면 sma-support@Snowflake.com으로 이메일을 보내주십시오. Snowflake와 지원 계약을 체결한 경우에는 영업 엔지니어에게 문의하여 지원 요구 사항을 안내받을 수 있습니다.

SPRKPY1041

경고

이 문제 코드는 Spark Conversion Core Version 2.9.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.functions.explode_outer에 해결 방법이 있습니다

카테고리: 경고

설명

이 문제는 도구가 pyspark.sql.functions.explode_outer 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제는 select 호출에서 explode_outer 메서드를 사용하는 방법을 보여줍니다.

df = spark.createDataFrame(
    [(1, ["foo", "bar"], {"x": 1.0}),
     (2, [], {}),
     (3, None, None)],
    ("id", "an_array", "a_map")
)

df.select("id", "an_array", explode_outer("a_map")).show()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1041 을 추가합니다.

df = spark.createDataFrame(
    [(1, ["foo", "bar"], {"x": 1.0}),
     (2, [], {}),
     (3, None, None)],
    ("id", "an_array", "a_map")
)

#EWI: SPRKPY1041 => pyspark.sql.functions.explode_outer has a workaround, see documentation for more info
df.select("id", "an_array", explode_outer("a_map")).show()
Copy

권장 수정

해결 방법으로 explode_outer 함수에 대한 헬퍼가 포함된 snowpark_extensions 패키지를 가져오면 됩니다.

import snowpark_extensions

df = spark.createDataFrame(
    [(1, ["foo", "bar"], {"x": 1.0}),
     (2, [], {}),
     (3, None, None)],
    ("id", "an_array", "a_map")
)

df.select("id", "an_array", explode_outer("a_map")).show()
Copy

추가 권장 사항

SPRKPY1075

카테고리

경고.

설명

Parse_json은 스키마 유효성 검사를 적용하지 않으므로 스키마에 따라 필터링/검증해야 하는 경우 몇 가지 논리를 도입해야 할 수 있습니다.

예

입력

df.select(from_json(df.value, Schema))
df.select(from_json(schema=Schema, col=df.value))
df.select(from_json(df.value, Schema, option))
Copy

출력

#EWI: SPRKPY1075 => The parse_json does not apply schema validation, if you need to filter/validate based on schema you might need to introduce some logic.
df.select(parse_json(df.value))
#EWI: SPRKPY1075 => The parse_json does not apply schema validation, if you need to filter/validate based on schema you might need to introduce some logic.
df.select(parse_json(df.value))
#EWI: SPRKPY1075 => The parse_json does not apply schema validation, if you need to filter/validate based on schema you might need to introduce some logic.
df.select(parse_json(df.value))
Copy

함수 from_json의 경우 스키마는 실제로 추론을 위해 전달되지 않고 유효성 검사에 사용됩니다. 이 예시를 참조하십시오.

data = [
    ('{"name": "John", "age": 30, "city": "New York"}',),
    ('{"name": "Jane", "age": "25", "city": "San Francisco"}',)
]

df = spark.createDataFrame(data, ["json_str"])
Copy

예 1: 데이터 타입 적용 및 열 이름 변경:

## Parse JSON column with schema
parsed_df = df.withColumn("parsed_json", from_json(col("json_str"), schema))

parsed_df.show(truncate=False)

## +------------------------------------------------------+---------------------------+
## |json_str                                              |parsed_json                |
## +------------------------------------------------------+---------------------------+
## |{"name": "John", "age": 30, "city": "New York"}       |{John, 30, New York}       |
## |{"name": "Jane", "age": "25", "city": "San Francisco"}|{Jane, null, San Francisco}|
## +------------------------------------------------------+---------------------------+
## notice that values outside of the schema were dropped and columns not matched are returned as null
Copy

예 2: 특정 열 선택:

## Define a schema with only the columns we want to use
partial_schema = StructType([
    StructField("name", StringType(), True),
    StructField("city", StringType(), True)
])

## Parse JSON column with partial schema
partial_df = df.withColumn("parsed_json", from_json(col("json_str"), partial_schema))

partial_df.show(truncate=False)

## +------------------------------------------------------+---------------------+
## |json_str                                              |parsed_json          |
## +------------------------------------------------------+---------------------+
## |{"name": "John", "age": 30, "city": "New York"}       |{John, New York}     |
## |{"name": "Jane", "age": "25", "city": "San Francisco"}|{Jane, San Francisco}|
## +------------------------------------------------------+---------------------+
## there is also an automatic filtering
Copy

권장 사항

  • 자세한 지원이 필요하면 sma-support@Snowflake.com으로 이메일을 보내주십시오. Snowflake와 지원 계약을 맺은 경우 영업 엔지니어에게 문의하면 지원 요구 사항을 안내해 드릴 수 있습니다.

  • 유용한 도구 PEP-8 및 다시 들여쓰기.

SPRKPY1024

메시지: pyspark.sql.functions.log2에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.log2 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 이 EWI 를 생성하는 pyspark.sql.functions.log2 함수의 사용 예제입니다. 이 예제에서는 log2 함수를 사용하여 value 열의 기본 2 로그 값을 계산합니다.

df = spark.createDataFrame([(1,), (2,), (4,), (8,), (16,)], ["value"])
df_with_log2 = df.withColumn("log2_value", log2(df["value"]))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1024 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([(1,), (2,), (4,), (8,), (16,)], ["value"])
#EWI: SPRKPY1024 => pyspark.sql.functions.log2 has a workaround, see documentation for more info
df_with_log2 = df.withColumn("log2_value", log2(df["value"]))
Copy

권장 수정

해결 방법으로 리터럴 값 2 를 기본값으로 전달하여 Snowflake.snowpark.functions.log 함수를 사용할 수 있습니다.

df = session.createDataFrame([(1,), (2,), (4,), (8,), (16,)], ["value"])
df_with_log2 = df.withColumn("log2_value", log(2, df["value"]))
Copy

추가 권장 사항

SPRKPY1086

메시지: pyspark.ml.linalg.VectorUDT 가 지원되지 않습니다.

카테고리: 경고

설명

pyspark.ml.linalg.VectorUDT 는 지원되지 않습니다.

시나리오

입력 코드

VectorUDT 는 DataFrame 에서 벡터 열을 나타내는 데이터 타입입니다.

data = [
        (1, Vectors.dense([10.0, 20.0])),
        (2, Vectors.dense([25.0, 30.0])),
        (3, Vectors.dense([50.0, 60.0]))
    ]

schema = StructType([
        StructField("Id", IntegerType(), True),
        StructField("VectorCol", VectorUDT(), True),
    ])

df = SparkSession.createDataFrame(data, schema=schema)
Copy

출력 코드

도구는 출력 코드에 EWI SPRKPY1086 을 추가하여 이 함수가 Snowpark에서 지원되지 않음을 알려줍니다.

data = [
        (1, Vectors.dense([10.0, 20.0])),
        (2, Vectors.dense([25.0, 30.0])),
        (3, Vectors.dense([50.0, 60.0]))
    ]

#EWI: SPRKPY1086 => The pyspark.ml.linalg.VectorUDT function is not supported.
schema = StructType([
        StructField("Id", IntegerType(), True),
        StructField("VectorCol", VectorUDT(), True),
    ])

df = spark.createDataFrame(data, schema=schema)
Copy

권장 수정

VectorUDT 에는 권장 수정 사항이 없습니다.

추가 권장 사항

SPRKPY1034

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.desc에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.desc 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

pyspark.sql.functions.desc 함수는 열 오브젝트 또는 열의 이름을 문자열로 매개 변수로 사용합니다. 두 시나리오 모두 Snowpark에서 지원하지 않으므로 EWI 가 생성됩니다.

시나리오 1

입력

다음은 열 오브젝트를 매개 변수로 사용하는 pyspark.sql.functions.desc 함수의 사용 예입니다.

df.orderBy(desc(col))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1034 를 추가하여 열 오브젝트 매개 변수가 있는 desc 함수가 Snowpark에서 직접 지원되지 않지만 우회 방법이 있음을 알려줍니다.

#EWI: SPRKPY1034 => pyspark.sql.functions.desc has a workaround, see documentation for more info
df.orderBy(desc(col))
Copy

권장 수정

해결 방법으로 column 매개 변수에서 Snowflake.snowpark.Column.desc 함수를 호출할 수 있습니다.

df.orderBy(col.desc())
Copy
시나리오 2

입력

다음은 열 이름을 매개 변수로 사용하는 pyspark.sql.functions.desc 함수의 사용 예입니다.

df.orderBy(desc("colName"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1034 를 추가하여 열 이름 매개 변수가 있는 desc 함수가 Snowpark에서 직접 지원되지 않지만 우회 방법이 있음을 알려줍니다.

#EWI: SPRKPY1034 => pyspark.sql.functions.desc has a workaround, see documentation for more info
df.orderBy(desc("colName"))
Copy

권장 수정

해결 방법으로 Snowflake.snowpark.functions.col 함수를 사용하여 문자열 매개 변수를 열 오브젝트로 변환한 다음 Snowflake.snowpark.Column.desc 함수를 호출하면 됩니다.

df.orderBy(col("colName").desc())
Copy

추가 권장 사항

SPRKPY1065

메시지: Snowflake는 데이터 클러스터링 메커니즘을 사용하여 데이터를 계산하기 때문에 pyspark.context.SparkContext.broadcast는 적용되지 않습니다.

카테고리: 경고

설명

이 문제는 Snowflake의 data-clustering 사용으로 인해 필요하지 않은 요소 pyspark.context.SparkContext.broadcast 의 사용이 툴에 의해 감지될 때 발생합니다.

입력 코드

이 예제에서는 브로드캐스트 변수가 생성되며, 이 변수를 통해 모든 노드에서 데이터를 보다 효율적으로 공유할 수 있습니다.

sc = SparkContext(conf=conf_spark)

mapping = {1: 10001, 2: 10002}

bc = sc.broadcast(mapping)
Copy

출력 코드

SMA 는 브로드캐스트가 필요하지 않음을 나타내는 EWI 메시지를 추가합니다.

sc = conf_spark

mapping = {1: 10001, 2: 10002}
#EWI: SPRKPY1065 => The element does not apply since snowflake use data-clustering mechanism to compute the data.

bc = sc.broadcast(mapping)
Copy

권장 수정

SparkContext.broadcast.

sc = conf_spark

mapping = {1: 10001, 2: 10002}
Copy

추가 권장 사항

SPRKPY1051

경고

이 문제 코드는 Spark Conversion Core Version 2.4.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.세션.SparkSession.Builder.master에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.session.SparkSession.Builder.master 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

다음은 builder.master 를 사용하여 1개 코어를 사용해 로컬에 연결하도록 Spark Master URL 을 설정하는 방법의 사용 예입니다.

spark = SparkSession.builder.master("local[1]")
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1051 를 추가합니다.

#EWI: SPRKPY1051 => pyspark.sql.session.SparkSession.Builder.master has a workaround, see documentation for more info
spark = Session.builder.master("local[1]")
Copy

권장 수정

SparkSession.Builder.master 는 Spark 클러스터를 설정하는 데 사용됩니다. Snowpark는 Spark 클러스터를 사용하지 않으므로 코드를 제거하거나 설명할 수 있습니다.

## spark = Session.builder.master("local[1]")
Copy

추가 권장 사항

SPRKPY1000

메시지: 소스 프로젝트 Spark-core 버전은 xx.xx:xx.x.x이고, Snowpark에서 지원하는 Spark-core 버전은 2.12:3.1.2이므로 기존 매핑과 함수 차이가 있을 수 있습니다.

카테고리: 경고.

설명

이 문제는 소스 코드의 Pyspark 버전이 지원되지 않을 때 표시됩니다. 즉, 기존 매핑 간에 기능적인 차이가 있을 수 있습니다.

추가 권장 사항

  • SMA 가 Snowpark와의 호환성을 위해 검사한 pyspark 버전은 2.12에서 3.1.2입니다. 이 범위를 벗어난 버전을 사용하는 경우 도구에서 일관성 없는 결과가 나올 수 있습니다. 스캔하는 소스 코드의 버전을 변경할 수 있습니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 으로 이메일을 보내거나 SMA 에 문제를 게시할 수 있습니다.

SPRKPY1081

이 문제 코드는 Spark Conversion Core 4.12.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.readwriter.DataFrameWriter.partitionBy에 해결 방법이 있습니다.

카테고리: 경고

설명

DataFrameWriter.partitionBy 함수는 지원되지 않습니다. 해결 방법은 Snowpark의 copy_into_location 을 대신 사용하는 것입니다. 자세한 내용은 설명서를 참조하십시오.

시나리오

입력

이 코드는 FIRST_NAME 열의 각 고유 값에 대해 별도의 디렉터리를 생성합니다. 데이터는 동일하지만 열에 따라 다른 디렉터리에 저장됩니다.

df = session.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]], schema = ["FIRST_NAME", "LAST_NAME"])
df.write.partitionBy("FIRST_NAME").csv("/home/data")
Copy

이 코드는 FIRST_NAME 열의 각 고유 값에 대해 별도의 디렉터리를 생성합니다. 데이터는 동일하지만 열에 따라 다른 디렉터리에 저장됩니다.

출력 코드

df = session.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]], schema = ["FIRST_NAME", "LAST_NAME"])
#EWI: SPRKPY1081 => The partitionBy function is not supported, but you can instead use copy_into_location as workaround. See the documentation for more info.
df.write.partitionBy("FIRST_NAME").csv("/home/data", format_type_options = dict(compression = "None"))
Copy

권장 수정

Snowpark에서 copy_into_location 에는 partitionBy 함수 대신 사용할 수 있는 partition_by 매개 변수가 있지만 다음 예제와 같이 몇 가지 수동 조정이 필요합니다.

Spark 코드:

df = session.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]], schema = ["FIRST_NAME", "LAST_NAME"])
df.write.partitionBy("FIRST_NAME").csv("/home/data")
Copy

Snowpark 코드를 수동으로 조정:

df = session.createDataFrame([["John", "Berry"], ["Rick", "Berry"], ["Anthony", "Davis"]], schema = ["FIRST_NAME", "LAST_NAME"])
df.write.copy_into_location(location=temp_stage, partition_by=col("FIRST_NAME"), file_format_type="csv", format_type_options={"COMPRESSION": "NONE"}, header=True)
Copy

copy_into_location 에는 다음 매개 변수가 있습니다

  • location: Snowpark 위치는 Snowflake 스테이지 를 사용하는 클라우드 위치만 허용합니다.

  • partition_by: 열 이름 또는 SQL 식일 수 있으므로 col 또는 sql_expr을 사용하여 열 또는 SQL 로 변환해야 합니다.

추가 권장 사항

SPRKPY1072

메시지: Snowpark에서 StorageLevel 사용은 필수가 아닙니다.

카테고리: 경고.

설명

이 문제는 도구가 저장소 레벨을 설정하는 “플래그”와 같은 역할을 하는 StorageLevel 클래스의 사용을 발견했을 때 발생합니다. Snowflake가 저장소를 제어하므로 이 함수를 사용할 필요가 없습니다.

추가 권장 사항

  • 이 함수의 모든 사용을 제거합니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1023

메시지: pyspark.sql.functions.log1p에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.log1p 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.functions.log1p 함수의 사용 예제입니다. 이 예제에서는 log1p 함수를 사용하여 value 열의 자연 로그를 계산합니다.

df = spark.createDataFrame([(0,), (1,), (10,), (100,)], ["value"])
df_with_log1p = df.withColumn("log1p_value", log1p(df["value"]))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1023 을 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([(0,), (1,), (10,), (100,)], ["value"])
#EWI: SPRKPY1023 => pyspark.sql.functions.log1p has a workaround, see documentation for more info
df_with_log1p = df.withColumn("log1p_value", log1p(df["value"]))
Copy

권장 수정

해결 방법으로 첫 번째 인자로 ln 문자열을 전달하고 두 번째 인자에 1 을 추가하여 call_function 함수를 사용할 수 있습니다.

df = spark.createDataFrame([(0,), (1,), (10,), (100,)], ["value"])
df_with_log1p = df.withColumn("log1p_value", call_function("ln", lit(1) + df["value"]))
Copy

추가 권장 사항

SPRKPY1017

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

pyspark.sql.functions.date_add에는 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.date_add 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제에서는 date_add를 사용하여 데이터프레임 df의 현재 날짜로부터 5일 후 날짜를 계산합니다.

col = df.select(date_add(df.colName, 5))
Copy

출력

SMA 는 date_add가 사용되는 라인 위에 EWI SPRKPY1017 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

#EWI: SPRKPY1017 => pyspark.sql.functions.date_add has a workaround, see documentation for more info
col = df.select(date_add(df.colName, 5))
Copy

권장 수정

Date_add (및 별칭 dateAdd) 함수에 대한 구현이 포함된 Snowflake.snowpark.functions를 가져옵니다.

from snowflake.snowpark.functions import date_add

col = df.select(date_add(df.dt, 1))
Copy

추가 권장 사항

SPRKPY1046

경고

이 문제 코드는 Spark Conversion Core Version 2.1.22 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.functions.monotonically_increasing_id에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.monotonically_increasing_id 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

다음은 monotonically_increasing_id 메서드의 사용 예시입니다.

from pyspark.sql import functions as F

spark.range(0, 10, 1, 2).select(F.monotonically_increasing_id()).show()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1046 을 추가합니다.

from pyspark.sql import functions as F
#EWI: SPRKPY1046 => pyspark.sql.functions.monotonically_increasing_id has a workaround, see documentation for more info
spark.range(0, 10, 1, 2).select(F.monotonically_increasing_id()).show()
Copy

권장 수정

도구 버전을 업데이트합니다.

추가 권장 사항

SPRKPY1056

경고

이 문제 코드는 사용 중단 되었습니다

메시지: DataFrameReader.option 인자 _ <argument_name> _는 리터럴이 아니므로 평가할 수 없습니다

카테고리: 경고

설명

이 문제는 pyspark.sql.readwriter.DataFrameReader.option 함수의 인자 키 또는 값이 리터럴 값이 아닌 경우(예: 변수)에 표시됩니다. SMA 는 소스 코드의 정적 분석을 수행하므로 인자의 내용을 평가할 수 없습니다.

시나리오

입력

다음은 pyspark.sql.readwriter.DataFrameReader.option 함수를 사용하여 EWI 를 생성하는 예제입니다.

my_value = ...
my_option = ...

df1 = spark.read.option("dateFormat", my_value).format("csv").load('filename.csv')
df2 = spark.read.option(my_option, "false").format("csv").load('filename.csv')
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1056 을 추가하여 이 함수의 인자가 리터럴 값이 아니므로 SMA 가 평가할 수 없음을 알립니다.

my_value = ...
my_option = ...

#EWI: SPRKPY1056 => pyspark.sql.readwriter.DataFrameReader.option argument "dateFormat" is not a literal and can't be evaluated
df1 = spark.read.option("dateFormat", my_value).format("csv").load('filename.csv')
#EWI: SPRKPY1056 => pyspark.sql.readwriter.DataFrameReader.option argument key is not a literal and can't be evaluated
df2 = spark.read.option(my_option, "false").format("csv").load('filename.csv')
Copy

권장 수정

SMA 가 해당 인자를 평가할 수 없었지만, 이는 Snowpark에서 지원하지 않는다는 것을 의미하지는 않습니다. 설명서 를 확인하여 인자의 값이 유효하고 Snowpark에서 동등한지 확인하십시오.

추가 권장 사항

SPRKPY1007

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

메시지 : pyspark.sql.context.SQLContext 는 필수가 아닙니다

카테고리 : 경고.

설명

이 문제는 도구가 필수가 아닌 pyspark.sql.context.SQLContext 의 사용을 감지할 때 발생합니다.

시나리오

입력

SparkContext 오버로드가 다른 예제는 다음과 같습니다.

from pyspark import SQLContext
​
my_sc1 = SQLContext(myMaster, myAppName, mySparkHome, myPyFiles, myEnvironment, myBatctSize, mySerializer, my_conf1)
my_sc2 = SQLContext(conf=my_conf2)
my_sc3 = SQLContext()
Copy

출력

출력 코드는 pyspark.SQLContext 에 대한 라인을 설명하고 시나리오를 구성에 대한 참조로 대체합니다. Spark 속성이 포함된 my_sc1 및 my_sc2 변수는 필수가 아니거나 코드를 수정하기 위해 변경해야 할 수 있습니다.

#EWI: SPRKPY1007 => pyspark.sql.context.SQLContext is not required
#from pyspark import SQLContext
pass

#EWI: SPRKPY1007 => pyspark.sql.context.SQLContext is not required
sql_context1 = my_sc1
#EWI: SPRKPY1007 => pyspark.sql.context.SQLContext is not required
sql_context2 = my_sc2
Copy

추가 권장 사항

  • 이는 불필요한 매개 변수이며 소스 코드에 경고 설명이 삽입되어 제거됩니다. 사용자의 작업이 없어야 합니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1033

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.asc에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.asc 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

pyspark.sql.functions.asc 함수는 열 오브젝트 또는 열의 이름을 문자열로 매개 변수로 사용합니다. 두 시나리오 모두 Snowpark에서 지원하지 않으므로 EWI 가 생성됩니다.

시나리오 1

입력

다음은 열 오브젝트를 매개 변수로 사용하는 pyspark.sql.functions.asc 함수의 사용 예입니다.

df.orderBy(asc(col))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1033 을 추가하여 열 오브젝트 매개 변수가 있는 asc 함수가 Snowpark에서 직접 지원되지 않지만 우회 방법이 있음을 알려줍니다.

#EWI: SPRKPY1033 => pyspark.sql.functions.asc has a workaround, see documentation for more info
df.orderBy(asc(col))
Copy

권장 수정

해결 방법으로 column 매개 변수에서 Snowflake.snowpark.Column.asc 함수를 호출할 수 있습니다.

df.orderBy(col.asc())
Copy
시나리오 2

입력

다음은 열 이름을 매개 변수로 사용하는 pyspark.sql.functions.asc 함수의 사용 예입니다.

df.orderBy(asc("colName"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1033 을 추가하여 열 이름 매개 변수가 있는 asc 함수가 Snowpark에서 직접 지원되지 않지만 우회 방법이 있음을 알려줍니다.

#EWI: SPRKPY1033 => pyspark.sql.functions.asc has a workaround, see documentation for more info
df.orderBy(asc("colName"))
Copy

권장 수정

해결 방법으로 Snowflake.snowpark.functions.col 함수를 사용하여 문자열 매개 변수를 열 오브젝트로 변환한 다음 Snowflake.snowpark.Column.asc 함수를 호출할 수 있습니다.

df.orderBy(col("colName").asc())
Copy

추가 권장 사항

SPRKPY1062

메시지: Snowpark는 매개 변수 “values”가 없는 GroupedData.pivot을 지원하지 않습니다.

카테고리: 경고

설명

이 문제는 SMA 가 “values” 매개 변수 (피벗할 값의 목록) 없이 pyspark.sql.group.GroupedData.pivot 함수의 사용을 감지할 때 발생합니다.

현재 Snowpark Python 피벗 함수를 사용하려면 피벗할 고유 값 목록을 명시적으로 지정해야 합니다.

시나리오

시나리오 1

입력

SMA 가 dataFrame.groupBy("columnX").pivot("columnY") 패턴과 일치하는 식을 감지하고 피벗에 values 매개 변수가 없습니다.

df.groupBy("date").pivot("category").sum("amount")
Copy

출력

SMA 는 “values” 매개 변수가 없는 피벗 함수가 지원되지 않음을 나타내는 EWI 메시지를 추가합니다.

또한 피벗 함수의 두 번째 매개 변수로 열로 변환될 값의 목록을 계산하는 목록 이해도를 추가합니다. 이 작업은 대규모 데이터 세트에는 효율적이지 않으므로 값을 명시적으로 표시하는 것이 좋습니다.

#EWI: SPRKPY1062 => pyspark.sql.group.GroupedData.pivot without parameter 'values' is not supported. See documentation for more info.
df.groupBy("date").pivot("category", [v[0] for v in df.select("category").distinct().limit(10000).collect()]]).sum("amount")
Copy

권장 수정

이 시나리오에서는 SMA 피벗 함수의 두 번째 매개 변수에 열로 변환될 값의 목록을 계산하는 목록 이해력을 추가하지만 다음과 같이 피벗할 고유 값의 목록을 지정할 수 있습니다.

df = spark.createDataFrame([
      Row(category="Client_ID", date=2012, amount=10000),
      Row(category="Client_name",   date=2012, amount=20000)
  ])

df.groupBy("date").pivot("category", ["dotNET", "Java"]).sum("amount")
Copy
시나리오 2

입력

SMA 가 dataFrame.groupBy("columnX").pivot("columnY") 패턴과 일치하는 식을 감지할 수 없으며 피벗에 values 매개 변수가 없습니다.

df1.union(df2).groupBy("date").pivot("category").sum("amount")
Copy

출력

SMA 는 “values” 매개 변수가 없는 피벗 함수가 지원되지 않음을 나타내는 EWI 메시지를 추가합니다.

#EWI: SPRKPY1062 => pyspark.sql.group.GroupedData.pivot without parameter 'values' is not supported. See documentation for more info.
df1.union(df2).groupBy("date").pivot("category").sum("amount")
Copy

권장 수정

다음과 같이 피벗할 고유 값 목록을 추가합니다.

df = spark.createDataFrame([
      Row(course="dotNET", year=2012, earnings=10000),
      Row(course="Java",   year=2012, earnings=20000)
  ])

df.groupBy("year").pivot("course", ["dotNET", "Java"]).sum("earnings").show()
Copy

추가 권장 사항

  • 피벗할 고유 값의 목록을 계산하는 것은 대규모 데이터 세트에서 효율적인 작업이 아니며 차단 호출이 될 수 있습니다. 피벗할 고유 값 목록을 명시적으로 표시하는 것을 고려해 보십시오.

  • 피벗할 고유 값 목록을 명시적으로 지정하지 않으려면(권장하지 않음), 다음 코드를 피벗 함수의 두 번째 인자로 추가하여 런타임에 값을 유추할 수 있습니다.*

[v[0] for v in <df>.select(<column>).distinct().limit(<count>).collect()]]
Copy

*<df> 를 해당 DataFrame 으로 _ 바꾸고 _ 피벗할 열과 선택할 행 수를 입력합니다.

SPRKPY1042

메시지: pyspark.sql.functions.posexplode에 해결 방법이 있습니다

카테고리: 경고

설명

이 문제는 도구가 pyspark.sql.functions.posexplode 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

이 메서드가 매개 변수로 전달되는 열의 유형에 따라 처리할 수 있는 몇 가지 시나리오가 있는데, list of values 또는 map/directory (keys/values) 일 수 있습니다.

시나리오 1

입력

list of values 의 매개 변수로 전달되는 posexplode 의 사용 예는 다음과 같습니다.

df = spark.createDataFrame(
    [Row(a=1,
         intlist=[1, 2, 3])])

df.select(posexplode(df.intlist)).collect()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1042 를 추가합니다.

df = spark.createDataFrame(
    [Row(a=1,
         intlist=[100, 200, 300])])
#EWI: SPRKPY1042 => pyspark.sql.functions.posexplode has a workaround, see documentation for more info

df.select(posexplode(df.intlist)).show()
Copy

권장 수정

동일한 동작을 하려면 functions.flatten 함수를 사용하고, 추가 열을 삭제하고, 인덱스 및 값 열 이름을 바꾸면 됩니다.

df = spark.createDataFrame(
  [Row(a=1,
       intlist=[1, 2, 3])])

df.select(
    flatten(df.intlist))\
    .drop("DATA", "SEQ", "KEY", "PATH", "THIS")\
    .rename({"INDEX": "pos", "VALUE": "col"}).show()
Copy
시나리오 2

입력

맵/사전(키/값) 을 매개 변수로 전달하는 posexplode 의 사용 예는 다음과 같습니다.

df = spark.createDataFrame([
    [1, [1, 2, 3], {"Ashi Garami": "Single Leg X"}, "Kimura"],
    [2, [11, 22], {"Sankaku": "Triangle"}, "Coffee"]
],
schema=["idx", "lists", "maps", "strs"])

df.select(posexplode(df.maps)).show()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1042 를 추가합니다.

df = spark.createDataFrame([
    [1, [1, 2, 3], {"Ashi Garami": "Single Leg X"}, "Kimura"],
    [2, [11, 22], {"Sankaku": "Triangle"}, "Coffee"]
],
schema=["idx", "lists", "maps", "strs"])
#EWI: SPRKPY1042 => pyspark.sql.functions.posexplode has a workaround, see documentation for more info

df.select(posexplode(df.maps)).show()
Copy

권장 수정

해결 방법으로 functions.row_number 를 사용하여 위치를 가져오고 functions.explode 를 필드 이름과 함께 사용하여 사전의 키/값을 가져오면 됩니다.

df = spark.createDataFrame([
    [10, [1, 2, 3], {"Ashi Garami": "Single Leg X"}, "Kimura"],
    [11, [11, 22], {"Sankaku": "Triangle"}, "Coffee"]
],
    schema=["idx", "lists", "maps", "strs"])

window = Window.orderBy(col("idx").asc())

df.select(
    row_number().over(window).alias("pos"),
    explode(df.maps).alias("key", "value")).show()
Copy

참고: row_number를 사용하는 것은 1로 시작하기 때문에 완전히 동일하지 않습니다(Spark 메서드처럼 0이 아님)

추가 권장 사항

SPRKPY1013

메시지: pyspark.sql.functions.acosh에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.acosh 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제에서 pyspark는 pyspark.sql.functions.acosh 를 사용하여 데이터 프레임에 대한 acosh를 계산합니다

from pyspark.sql import SparkSession
from pyspark.sql.functions import acosh
spark = SparkSession.builder.getOrCreate()
data = [['V1', 30],
        ['V2', 60],
        ['V3', 50],
        ['V4', 13]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
df_with_acosh = df.withColumn("acosh_value", acosh(df["value"]))
Copy

출력

SMA 는 acosh가 사용되는 라인 위에 EWI SPRKPY1013 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

from snowflake.snowpark import Session

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['V1', 30],
        ['V2', 60],
        ['V3', 50],
        ['V4', 13]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
#EWI: SPRKPY1013 => pyspark.sql.functions.acosh has a workaround, see documentation for more info
df_with_acosh = df.withColumn("acosh_value", acosh(df["value"]))
Copy

권장 수정

직접적인 “acosh”구현은 없지만 “call_function”을 대신 사용할 수 있으며, 첫 번째 매개 변수로 “acosh”를 사용하고 두 번째 매개 변수로 colName을 사용할 수 있습니다.

import snowflake.snowpark as snowpark
from snowflake.snowpark import Session
from snowflake.snowpark.functions import call_function, col

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['V1', 30],
        ['V2', 60],
        ['V3', 50],
        ['V4', 13]]

columns = ['Paremeter', 'value']
df = spark.createDataFrame(data, columns)
df_with_acosh = df.select(call_function('ACOSH', col('value')))
Copy

추가 권장 사항

SPRKPY1085

메시지: pyspark.ml.feature.VectorAssembler 는 지원되지 않습니다.

카테고리: 경고

설명

pyspark.ml.feature.VectorAssembler 는 지원되지 않습니다.

시나리오

입력 코드

VectorAssembler 는 여러 열을 단일 벡터로 결합하는 데 사용됩니다.

data = [
        (1, 10.0, 20.0),
        (2, 25.0, 30.0),
        (3, 50.0, 60.0)
    ]

df = SparkSession.createDataFrame(data, schema=["Id", "col1", "col2"])
vector = VectorAssembler(inputCols=["col1", "col2"], output="cols")
Copy

출력 코드

도구는 출력 코드에 EWI SPRKPY1085 를 추가하여 이 클래스가 Snowpark에서 지원되지 않음을 알려줍니다.

data = [
        (1, 10.0, 20.0),
        (2, 25.0, 30.0),
        (3, 50.0, 60.0)
    ]

df = spark.createDataFrame(data, schema=["Id", "col1", "col2"])
#EWI: SPRKPY1085 => The pyspark.ml.feature.VectorAssembler function is not supported.

vector = VectorAssembler(inputCols=["col1", "col2"], output="cols")
Copy

권장 수정

pyspark.ml.feature.VectorAssembler 에는 권장 수정 사항이 없습니다.

추가 권장 사항

SPRKPY1027

경고

이 문제 코드는 Spark Conversion Core 4.5.2 이후에 사용 중단 되었습니다.

메시지: DataFrameReader.json에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.readwriter.DataFrameReader.json 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.readwriter.DataFrameReader.json 함수의 사용 예제입니다. 이 예제에서는 json 함수를 사용하여 지정된 스키마로 여러 개의 .json 파일을 읽고 primitiveAsString 및 dateFormat 같은 추가 옵션을 사용하여 파일 읽기 동작을 미세 조정합니다.

file_paths = [
  "path/to/your/file1.json",
  "path/to/your/file2.json",
  "path/to/your/file3.json",
]

df = session.read.json(
  file_paths,
  schema=my_schema,
  primitiveAsString=True,
  dateFormat="2023-06-20"
)
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1027 을 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

file_paths = [
  "path/to/your/file1.json",
  "path/to/your/file2.json",
  "path/to/your/file3.json",
]

#EWI: SPRKPY1027 => pyspark.sql.readwriter.DataFrameReader.json has a workaround, see documentation for more info
df = session.read.json(
  file_paths,
  schema=my_schema,
  primitiveAsString=True,
  dateFormat="2023-06-20"
)
Copy

권장 수정

이 섹션에서는 path 매개 변수, schema 매개 변수 및 일부 options 을 Snowpark에서 작동하도록 구성하는 방법에 대해 설명합니다.

1. path 매개 변수

Snowpark는 path 매개 변수를 스테이지 위치로 요구하므로 해결 방법으로 임시 스테이지를 생성하고 file:// 접두사를 사용하여 각 .json 파일을 해당 스테이지에 추가할 수 있습니다.

2. schema 매개 변수

Snowpark에서는 schema 를 json 함수의 매개 변수로 정의하는 것을 허용하지 않습니다. 해결 방법으로 snowflake.snowpark.DataFrameReader.schema 함수를 사용할 수 있습니다.

3. options 매개 변수

Snowpark는 extra options 을 json 함수의 매개 변수로 정의하는 것을 허용하지 않습니다. 해결 방법으로 많은 경우 DataFrameReader.option 함수를 사용하여 해당 매개 변수를 DataFrameReader 의 옵션으로 지정할 수 있습니다.

참고

다음 옵션은 Snowpark에서 지원되지 않음:

  • allowBackslashEscapingAnyCharacter

  • allowComments

  • allowNonNumericNumbers

  • allowNumericLeadingZero

  • allowSingleQuotes

  • allowUnquotedControlChars

  • allowUnquotedFieldNames

  • columnNameOfCorruptRecord

  • dropFiledIfAllNull

  • 인코딩

  • ignoreNullFields

  • lineSep

  • locale

  • 모드

  • multiline

  • prefersDecimal

  • primitiveAsString

  • samplingRatio

  • timestampNTZFormat

  • timeZone

아래는 위에서 언급한 제안을 적용하여 Snowpark에서 작동하도록 한 후 입력 코드의 전체 예시입니다.

stage = f'{session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {stage}')

session.file.put(f"file:///path/to/your/file1.json", f"@{stage}")
session.file.put(f"file:///path/to/your/file2.json", f"@{stage}")
session.file.put(f"file:///path/to/your/file3.json", f"@{stage}")

df = session.read.schema(my_schema).option("dateFormat", "2023-06-20").json(stage)
Copy

추가 권장 사항

SPRKPY1076

메시지: pyspark.sql.readwriter.DataFrameReader 메서드의 매개 변수는 지원되지 않습니다. 이는 CSV, JSON 및 PARQUET 메서드에 적용됩니다.

카테고리: 경고.

설명

pyspark.sql.readwriter.DataFrameReader 오브젝트에 대한 CSV, JSON 및 PARQUET 메서드의 경우 도구가 매개 변수를 분석하고 각 경우에 따라 변환을 추가합니다.

  • 모든 매개 변수는 Snowpark에서 해당 이름과 일치합니다. 이 경우 도구는 매개 변수를 .option() 호출로 변환합니다. 이 경우 매개 변수는 EWI 를 추가하지 않습니다.

  • 일부 매개 변수는 Snowpark의 해당 매개 변수와 일치하지 않습니다. 이 경우 도구는 EWI 매개 변수 정보를 추가하고 메서드 호출에서 제거합니다.

동등 매개 기능 목록:

  • CSV 동등 기능:

Spark 키

Snowpark 동등 기능

sep

FIELD_DELIMITER

헤더

PARSE_HEADER

lineSep

RECORD_DELIMITER

pathGlobFilter

PATTERN

인용

FIELD_OPTIONALLY_ENCLOSED_BY

nullValue

NULL_IF

날짜 형식

DATE_FORMAT

타임스탬프 형식

TIMESTAMP_FORMAT

추론 스키마

INFER_SCHEMA

구분_기호

FIELD_DELIMITER

  • JSON 동등 기능:

Spark 키

Snowpark 동등 기능

날짜 형식

DATE_FORMAT

타임스탬프 형식

TIMESTAMP_FORMAT

pathGlobFilter

PATTERN

  • PARQUET 동등 기능:

Spark 키

Snowpark 동등 기능

pathGlobFilter

PATTERN

시나리오

시나리오 1

입력

CVS 의 경우 몇 가지 예가 있습니다.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('myapp').getOrCreate()

spark.read.csv("path3", None,None,None,None,None,None,True).show()
Copy

출력

변환된 코드에서 매개 변수는 cvs 함수에 개별 옵션으로 추가됩니다

from snowflake.snowpark import Session

spark = Session.builder.app_name('myapp', True).getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})

#EWI: SPRKPY1076 => Some of the included parameters are not supported in the csv function, the supported ones will be added into a option method.
spark.read.option("FIELD_DELIMITER", None).option("PARSE_HEADER", True).option("FIELD_OPTIONALLY_ENCLOSED_BY", None).csv("path3").show()
Copy

시나리오 2

입력

JSON 의 경우 몇 가지 예가 있습니다.

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('myapp').getOrCreate()
spark.read.json("/myPath/jsonFile/", dateFormat='YYYY/MM/DD').show()
Copy

출력

변환된 코드에서 매개 변수는 json 함수에 개별 옵션으로 추가됩니다

from snowflake.snowpark import Session
spark = Session.builder.app_name('myapp', True).getOrCreate()
#EWI: SPRKPY1076 => Some of the included parameters are not supported in the json function, the supported ones will be added into a option method.

spark.read.option("DATE_FORMAT", 'YYYY/MM/DD').json("/myPath/jsonFile/").show()
Copy
시나리오 3

입력

PARQUET 의 경우 몇 가지 예가 있습니다.

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('myapp').getOrCreate()

spark.read.parquet("/path/to/my/file.parquet", pathGlobFilter="*.parquet").show()
Copy

출력

변환된 코드에서 매개 변수는 parquet 함수에 개별 옵션으로 추가됩니다

from snowflake.snowpark import Session

spark = Session.builder.app_name('myapp', True).getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})

#EWI: SPRKPY1076 => Some of the included parameters are not supported in the parquet function, the supported ones will be added into a option method.
#EWI: SPRKPY1029 => The parquet function require adjustments, in Snowpark the parquet files needs to be located in an stage. See the documentation for more info.

spark.read.option("PATTERN", "*.parquet").parquet("/path/to/my/file.parquet")
Copy

추가 권장 사항

SPRKPY1066

메시지: Snowflake 사용 마이크로 파티션 메커니즘은 자동으로 생성되므로 Spark 요소는 적용되지 않습니다.

카테고리: 경고

설명

이 문제는 도구가 파티션과 관련된 요소의 사용을 감지할 때 표시됩니다.

이러한 요소는 Snowflake의 마이크로 파티션 을 사용하기 때문에 적용되지 않습니다.

입력 코드

이 sortWithinPartitions 예제에서 이는 지정된 열을 기준으로 정렬된 DataFrame 에 파티션을 생성하는 데 사용됩니다.

df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"])
df.sortWithinPartitions("age", ascending=False)
Copy

출력 코드

SMA 에는 Spark 요소가 필수가 아님을 나타내는 EWI 메시지가 추가됩니다.

df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"])
#EWI: SPRKPY1066 => The element does not apply since snowflake use micro-partitioning mechanism are created automatically.
df.sortWithinPartitions("age", ascending=False)
Copy

권장 수정

요소의 사용법을 제거합니다.

df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"])
Copy

추가 권장 사항

SPRKPY1037

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.functions.sort_array에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.sort_array 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.functions.sort_array 함수의 사용 예제입니다. 이 예제에서는 sort_array 함수를 사용하여 숫자 배열을 오름차순 및 내림차순으로 정렬합니다.

df = spark.createDataFrame([(1, [3, 1, 2]), (2, [10, 5, 8]), (3, [6, 4, 7])], ["id", "numbers"])
df_sorted_asc = df.withColumn("sorted_numbers_asc", sort_array("numbers", asc=True))
df_sorted_desc = df.withColumn("sorted_numbers_desc", sort_array("numbers", asc=False))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1037 을 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([(1, [3, 1, 2]), (2, [10, 5, 8]), (3, [6, 4, 7])], ["id", "numbers"])
#EWI: SPRKPY1037 => pyspark.sql.functions.sort_array has a workaround, see documentation for more info
df_sorted_asc = df.withColumn("sorted_numbers_asc", sort_array("numbers", asc=True))
#EWI: SPRKPY1037 => pyspark.sql.functions.sort_array has a workaround, see documentation for more info
df_sorted_desc = df.withColumn("sorted_numbers_desc", sort_array("numbers", asc=False))
Copy

권장 수정

해결 방법으로 sort_array 함수에 대한 확장자를 제공하는 snowpark_extensions 패키지를 가져올 수 있습니다.

import snowpark_extensions

df = spark.createDataFrame([(1, [3, 1, 2]), (2, [10, 5, 8]), (3, [6, 4, 7])], ["id", "numbers"])
df_sorted_asc = df.withColumn("sorted_numbers_asc", sort_array("numbers", asc=True))
df_sorted_desc = df.withColumn("sorted_numbers_desc", sort_array("numbers", asc=False))
Copy

추가 권장 사항

SPRKPY1003

메시지 : 기호 테이블을 로딩하는 동안 오류가 발생했습니다.

카테고리 : 변환 오류입니다.

설명

이 문제는 기호 테이블에서 기호를 처리하는 데 오류가 있을 때 표시됩니다. 기호 테이블은 SMA 의 기본 아키텍처의 일부로, 보다 복잡한 변환을 가능하게 합니다. 이 오류는 소스 코드에 예기치 않은 문이 있기 때문일 수 있습니다.

추가 권장 사항

  • 이는 소스 코드 자체의 오류라기보다는 도구가 소스 코드를 처리하는 방법의 오류일 가능성이 높습니다. 가장 좋은 해결 방법은 SMA 에 문제를 게시하는 것입니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 으로 이메일을 보내거나 SMA 에 문제를 게시할 수 있습니다.

SPRKPY1052

경고

이 문제 코드는 Spark Conversion Core Version 2.8.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.session.SparkSession.Builder.enableHiveSupport 에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.session.SparkSession.Builder.enableHiveSupport 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

아래는 SparkSession 을 구성하고 enableHiveSupport 메서드를 사용하여 Hive 지원을 활성화하는 예제입니다.

spark = Session.builder.appName("Merge_target_table")\
        .config("spark.port.maxRetries","100") \
        .enableHiveSupport().getOrCreate()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1052 를 추가합니다.

#EWI: SPRKPY1052 => pyspark.sql.session.SparkSession.Builder.enableHiveSupport has a workaround, see documentation for more info
spark = Session.builder.appName("Merge_target_table")\
        .config("spark.port.maxRetries","100") \
        .enableHiveSupport().getOrCreate()
Copy

권장 수정

enableHiveSupport 함수는 Snowpark에서 필요하지 않으므로 사용을 제거합니다.

spark = Session.builder.appName("Merge_target_table")\
        .config("spark.port.maxRetries","100") \
        .getOrCreate()
Copy

추가 권장 사항

SPRKPY1043

메시지: pyspark.sql.functions.posexplode_outer에 해결 방법이 있습니다

카테고리: 경고

설명

이 문제는 도구가 pyspark.sql.functions.posexplode_outer 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

이 메서드가 매개 변수로 전달되는 열의 유형에 따라 처리할 수 있는 몇 가지 시나리오가 있는데, list of values 또는 map/directory (keys/values) 일 수 있습니다.

시나리오 1

입력

값 목록 을 전달하는 posexplode_outer 의 사용법을 보여주는 예는 다음과 같습니다.

df = spark.createDataFrame(
    [
        (1, ["foo", "bar"]),
        (2, []),
        (3, None)],
    ("id", "an_array"))

df.select("id", "an_array", posexplode_outer("an_array")).show()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1043 을 추가합니다.

df = spark.createDataFrame(
    [
        (1, ["foo", "bar"]),
        (2, []),
        (3, None)],
    ("id", "an_array"))
#EWI: SPRKPY1043 => pyspark.sql.functions.posexplode_outer has a workaround, see documentation for more info

df.select("id", "an_array", posexplode_outer("an_array")).show()
Copy

권장 수정

동일한 동작을 얻으려면 outer 매개 변수를 true로 보내고, 추가 열을 삭제하고, 인덱스 및 값 열 이름을 변경하는 functions.flatten 함수를 사용하십시오.

df = spark.createDataFrame(
    [
        (1, ["foo", "bar"]),
        (2, []),
        (3, None)],
    ("id", "an_array"))

df.select(
    flatten(df.an_array, outer=True))\
    .drop("DATA", "SEQ", "KEY", "PATH", "THIS")\
    .rename({"INDEX": "pos", "VALUE": "col"}).show()
Copy
시나리오 2

입력

map/dictionary (keys/values) 을 전달하는 posexplode_outer의 또 다른 사용 예는 다음과 같습니다.

df = spark.createDataFrame(
    [
        (1, {"x": 1.0}),
        (2, {}),
        (3, None)],
    ("id", "a_map"))

df.select(posexplode_outer(df.a_map)).show()
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1043 을 추가합니다.

df = spark.createDataFrame(
    [
        (1, {"x": "Ashi Garami"}),
        (2, {}),
        (3, None)],
    ("id", "a_map"))
#EWI: SPRKPY1043 => pyspark.sql.functions.posexplode_outer has a workaround, see documentation for more info

df.select(posexplode_outer(df.a_map)).show()
Copy

권장 수정

해결 방법으로 functions.row_number 를 사용하여 위치를 가져오고 functions.explode_outer 를 필드 이름과 함께 사용하여 사전의 키/값을 가져오면 됩니다.

df = spark.createDataFrame(
    [
        (1, {"x": "Ashi Garami"}),
        (2,  {}),
        (3, None)],
    ("id", "a_map"))

window = Window.orderBy(col("id").asc())

df.select(
    row_number().over(window).alias("pos"),
          explode_outer(df.a_map)).show()
Copy

참고: row_number를 사용하는 것은 1로 시작하기 때문에 완전히 동일하지 않습니다(Spark 메서드처럼 0이 아님)

추가 권장 사항

SPRKPY1012

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.dataframe.DataFrameStatFunctions.writeTo에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 ppyspark.sql.dataframe.DataFrameStatFunctions.writeTo 사용을 감지했을 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

이 예제에서는 데이터프레임 df가 Spark 테이블 “table”에 기록됩니다.

writer = df.writeTo("table")
Copy

출력

SMA 는 DataFrameStatFunctions.writeTo가 사용되는 라인 위에 EWI SPRKPY1012 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

#EWI: SPRKPY1012 => pyspark.sql.dataframe.DataFrameStatFunctions.writeTo has a workaround, see documentation for more info
writer = df.writeTo("table")
Copy

권장 수정

대신 df.write.SaveAsTable()를 사용합니다.

import df.write as wt
writer = df.write.save_as_table(table)
Copy

추가 권장 사항

SPRKPY1084

이 문제 코드는 Spark Conversion Core 4.12.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.readwriter.DataFrameWriter.option은 지원되지 않습니다.

카테고리: 경고

설명

DataFrameWriter.option 함수는 지원되지 않습니다.

시나리오

입력 코드

아래는 option 메서드를 사용한 예시이며, 이 메서드는 DataFrame 의 데이터를 작성할 때 추가 구성을 추가하는 데 사용됩니다.

path_csv_file = "/path/to/file.csv"
data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

df.write.option("header", True).csv(csv_file_path)
df.write.option("sep", ";").option("lineSep","-").csv(csv_file_path)
Copy

출력 코드

이 도구는 출력 코드에 EWI SPRKPY1084 를 추가하여 이 함수가 Snowpark에서 지원되지 않음을 알려줍니다.

path_csv_file = "/path/to/file.csv"
data = [
        ("John", 30, "New York"),
        ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])

#EWI: SPRKPY1084 => The pyspark.sql.readwriter.DataFrameWriter.option function is not supported.

df.write.option("header", True).csv(csv_file_path)
#EWI: SPRKPY1084 => The pyspark.sql.readwriter.DataFrameWriter.option function is not supported.
df.write.option("sep", ";").option("lineSep","-").csv(csv_file_path)
Copy

권장 수정

DataFrameWriter.option 메서드에는 권장 수정 사항이 없습니다.

추가 권장 사항

SPRKPY1026

경고

이 문제 코드는 Spark Conversion Core 4.3.2 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.readwriter.DataFrameReader.csv에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.readwriter.DataFrameReader.csv 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.readwriter.DataFrameReader.csv 함수의 사용 예제입니다. 이 예제에서는 csv 함수를 사용하여 지정된 스키마를 가진 여러 .csv 파일을 읽고 encoding, header 및 sep 같은 추가 옵션을 사용하여 파일 읽기 동작을 미세 조정합니다.

file_paths = [
  "path/to/your/file1.csv",
  "path/to/your/file2.csv",
  "path/to/your/file3.csv",
]

df = session.read.csv(
  file_paths,
  schema=my_schema,
  encoding="UTF-8",
  header=True,
  sep=","
)
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1026 을 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

file_paths = [
  "path/to/your/file1.csv",
  "path/to/your/file2.csv",
  "path/to/your/file3.csv",
]

#EWI: SPRKPY1026 => pyspark.sql.readwriter.DataFrameReader.csv has a workaround, see documentation for more info
df = session.read.csv(
  file_paths,
  schema=my_schema,
  encoding="UTF-8",
  header=True,
  sep=","
)
Copy

권장 수정

이 섹션에서는 path 매개 변수, schema 매개 변수 및 일부 options 을 Snowpark에서 작동하도록 구성하는 방법에 대해 설명합니다.

1. path 매개 변수

Snowpark는 path 매개 변수를 스테이지 위치로 요구하므로 해결 방법으로 임시 스테이지를 생성하고 file:// 접두사를 사용하여 각 .csv 파일을 해당 스테이지에 추가할 수 있습니다.

2. schema 매개 변수

Snowpark는 schema 를 csv 함수의 매개 변수로 정의하는 것을 허용하지 않습니다. 해결 방법으로 snowflake.snowpark.DataFrameReader.schema 함수를 사용할 수 있습니다.

3. options 매개 변수

Snowpark에서는 추가 옵션 을 csv 함수의 매개 변수로 정의하는 것을 허용하지 않습니다. 해결 방법으로 많은 경우 DataFrameReader.option 함수를 사용하여 해당 매개 변수를 DataFrameReader 의 옵션으로 지정할 수 있습니다.

참고

다음 옵션은 Snowpark에서 지원되지 않음:

  • columnNameOfCorruptRecord

  • emptyValue

  • enforceSchema

  • 헤더

  • ignoreLeadingWhiteSpace

  • ignoreTrailingWhiteSpace

  • 추론 스키마

  • locale

  • maxCharsPerColumn

  • maxColumns

  • 모드

  • multiLine

  • nanValue

  • negativeInf

  • nullValue

  • positiveInf

  • quoteAll

  • samplingRatio

  • timestampNTZFormat

  • unescapedQuoteHandling

아래는 위에서 언급한 제안을 적용하여 Snowpark에서 작동하도록 한 후 입력 코드의 전체 예시입니다.

stage = f'{session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {stage}')

session.file.put(f"file:///path/to/your/file1.csv", f"@{stage}")
session.file.put(f"file:///path/to/your/file2.csv", f"@{stage}")
session.file.put(f"file:///path/to/your/file3.csv", f"@{stage}")

df = session.read.schema(my_schema).option("encoding", "UTF-8").option("sep", ",").csv(stage)
Copy

추가 권장 사항

SPRKPY1077

메시지: SQL 임베디드 코드를 처리할 수 없습니다.

카테고리: 경고.

설명

이 문제는 도구에서 Snowpark로 변환할 수 없는 SQL 임베디드 코드를 감지할 때 표시됩니다.

자세한 내용은 SQL 임베디드 코드 섹션을 참조하십시오.

시나리오

입력

이 예제에서는 SQL 코드가 Pyspark.sql 메서드의 매개 변수로 사용되는 쿼리라는 변수에 포함되어 있습니다.

query = f"SELECT * from myTable"
spark.sql(query)
Copy

출력

SMA 는 PySpark.sql 매개 변수가 SQL 코드가 아닌 변수임을 감지하여 EWI SPRKPY1077 메시지가 PySpark.sql 라인에 추가됩니다.

query = f"SELECT * myTable"
#EWI: SPRKPY1077 => SQL embedded code cannot be processed.
spark.sql(query)
Copy

추가 권장 사항

  • SQL 의 변환을 위해 이 코드는 보간 없이 문자열 값으로만 메서드의 매개 변수로 직접 내부에 있어야 합니다. Snowflake에서 기능을 확인하려면 PySpark.SQL 함수로 보내기를 확인하십시오.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1067

메시지: Pyspark.sql.functions.split에는 Snowpark에서 지원되지 않는 매개 변수가 있습니다.

카테고리: 경고

설명

이 문제는 도구가 2개 이상의 매개 변수 또는 정규식 패턴을 매개 변수로 사용하는 pyspark.sql.functions.split 의 사용을 감지할 때 나타나며, 두 경우 모두 지원되지 않습니다.

시나리오

시나리오 1

입력 코드

이 예제에서 분할 함수에는 2개 이상의 매개 변수가 있습니다.

df.select(split(columnName, ",", 5))
Copy

출력 코드

도구는 출력 코드에 EWI 를 추가하여 2개 이상의 매개 변수가 있는 경우 이 함수가 지원되지 않음을 알려줍니다.

#EWI: SPRKPY1067 => Snowpark does not support split functions with more than two parameters or containing regex pattern. See documentation for more info.
df.select(split(columnName, ",", 5))
Copy

권장 수정

2개의 매개 변수만 사용하여 분할 함수를 유지합니다.

df.select(split(columnName, ","))
Copy
시나리오 2

입력 코드

이 예제에서 분할 함수에는 매개 변수로 정규식 패턴이 있습니다.

df.select(split(columnName, "^([\d]+-[\d]+-[\d])"))
Copy

출력 코드

이 도구는 출력 코드에 EWI 를 추가하여 이 함수가 매개 변수로 정규식 패턴이 있는 경우 지원되지 않음을 알려줍니다.

#EWI: SPRKPY1067 => Snowpark does not support split functions with more than two parameters or containing regex pattern. See documentation for more info.
df.select(split(columnName, "^([\d]+-[\d]+-[\d])"))
Copy

권장 수정

이 메서드 functions.split(str: ColumnOrName, pattern: str, limit: int = - 1) 에 대한 Spark 서명이 Snowpark의 functions.split(str: Union[Column, str], pattern: Union[Column, str]) so for now the scenario using regular expression does not hav 메서드와 정확히 일치하지 않습니다. 따라서 현재로서는 정규식을 사용하는 시나리오에 권장되는 수정 사항이 없습니다.

추가 권장 사항

SPRKPY1036

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.column.Column.getField에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.column.Column.getField 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.column.Column.getField 함수의 사용 예제입니다. 이 예제에서는 getField 함수를 사용하여 info 열에서 name 을 추출합니다.

df = spark.createDataFrame([(1, {"name": "John", "age": 30}), (2, {"name": "Jane", "age": 25})], ["id", "info"])
df_with_name = df.withColumn("name", col("info").getField("name"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1036 을 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([(1, {"name": "John", "age": 30}), (2, {"name": "Jane", "age": 25})], ["id", "info"])
#EWI: SPRKPY1036 => pyspark.sql.column.Column.getField has a workaround, see documentation for more info
df_with_name = df.withColumn("name", col("info").getField("name"))
Copy

권장 수정

해결 방법으로 필드 이름을 인덱스로 사용하여 Snowpark 열 인덱서 연산자 를 사용할 수 있습니다.

df = spark.createDataFrame([(1, {"name": "John", "age": 30}), (2, {"name": "Jane", "age": 25})], ["id", "info"])
df_with_name = df.withColumn("name", col("info")["name"])
Copy

추가 권장 사항

SPRKPY1002

메시지 : < 요소 > 지원되지 않음, Spark 요소 지원되지 않음.

카테고리 : 변환 오류입니다.

설명

이 문제는 도구가 Snowpark에서 지원되지 않는 요소의 사용을 감지하고 이와 관련된 자체 오류 코드가 없을 때 표시됩니다. 지원되지 않는 요소에 대해 SMA 에서 사용하는 일반적인 오류 코드입니다.

추가 권장 사항

  • 메시지의 옵션이나 요소가 지원되지 않는다고 해서 해결 방법을 찾을 수 없다는 의미는 아닙니다. 도구 자체에서 해결책을 찾을 수 없다는 의미일 뿐입니다.

  • Pyspark.ml 라이브러리에서 지원되지 않는 요소를 발견한 경우 몇 가지 다른 접근법을 고려해 보십시오. Snowflake에서 제공하는 이 가이드와 같이 ml과 관련된 문제를 해결하는 데 사용할 수 있는 추가 가이드가 있습니다.

  • 소스 코드에 올바른 구문이 있는지 확인합니다. (issues.csv 파일을 사용하여 변환 오류가 발생하는 위치를 확인할 수 있습니다.) 구문이 올바른 경우 SMA 에서 문제 신고 옵션을 사용하여 특정 요소에서 변환 오류가 발생했음을 신고하십시오. 이 문제를 파일로 제출할 때 설명에 오류를 일으킨 코드 라인을 포함하십시오.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1053

메시지: Dbc 파일을 추출하는 동안 오류가 발생했습니다.

카테고리: 경고.

설명

이 문제는 dbc 파일을 추출할 수 없을 때 표시됩니다. 이 경고는 다음 중 한 가지 이상의 이유로 인해 발생할 수 있습니다. 너무 무겁거나, 액세스할 수 없거나, 읽기 전용입니다.

추가 권장 사항

  • 해결 방법으로 파일이 너무 커서 처리할 수 없는 경우 파일 크기를 확인할 수 있습니다. 또한 도구가 액세스할 수 있는지 분석하여 액세스 문제를 방지하십시오.

  • 자세한 지원이 필요하면 이메일(snowconvert-info@Snowflake.com)로 문의하십시오. Snowflake와 지원 계약을 맺은 경우 영업 엔지니어에게 문의하면 지원 요구 사항을 안내해 드릴 수 있습니다.

SPRKPY1080

메시지: SparkContext 값은 ‘session’ 변수로 대체되었습니다.

카테고리: 경고

설명

Spark 컨텍스트는 Snowpark 세션을 생성하는 세션이라는 변수에 저장됩니다.

시나리오

입력

이 스니펫은 SparkContext 를 설명합니다.

## Input Code
from pyspark import SparkContext
from pyspark.sql import SparkSession

def example1():

    sc = SparkContext("local[*]", "TestApp")

    sc.setLogLevel("ALL")
    sc.setLogLevel("DEBUG")
Copy

출력

이 출력 코드에서 SMA 는 PySpark.SparkContext 를 SparkSession 로 바꿨습니다. SMA 도 “connection.json” 파일에 연결을 대체하는 템플릿을 추가한 다음 connection_parameter 변수에 이 구성을 로딩합니다.

## Output Code
import logging
import sys
import json
from snowflake.snowpark import Session
from snowflake.snowpark import Session

def example1():
    jsonFile = open("connection.json")
    connection_parameter = json.load(jsonFile)
    jsonFile.close()
    #EWI: SPRKPY1080 => The value of SparkContext is replaced with 'session' variable.
    sc = Session.builder.configs(connection_parameter).getOrCreate()
    sc.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
    logging.basicConfig(stream = sys.stdout, level = logging.NOTSET)
    logging.basicConfig(stream = sys.stdout, level = logging.DEBUG)
Copy

권장 수정

구성 파일 “connection.json”을 필수 연결 정보로 업데이트해야 합니다.

{
  "user": "my_user",
  "password": "my_password",
  "account": "my_account",
  "role": "my_role",
  "warehouse": "my_warehouse",
  "database": "my_database",
  "schema": "my_schema"
}
Copy

추가 권장 사항

SPRKPY1073

메시지: 매개 변수가 없거나 반환 유형 매개 변수가 없는 pyspark.sql.functions.udf는 지원되지 않습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.udf 를 함수 또는 데코레이터로 사용하는 것을 감지하고 매개 변수나 반환 유형 매개 변수가 없는 두 가지 특정 경우에 지원되지 않을 때 나타납니다.

시나리오

시나리오 1

입력

Pyspark에서는 입력 또는 반환 유형 매개 변수 없이 사용자 정의 함수를 생성할 수 있습니다.

from pyspark.sql import SparkSession, DataFrameStatFunctions
from pyspark.sql.functions import col, udf

spark = SparkSession.builder.getOrCreate()
data = [['Q1', 'Test 1'],
        ['Q2', 'Test 2'],
        ['Q3', 'Test 1'],
        ['Q4', 'Test 1']]

columns = ['Quadrant', 'Value']
df = spark.createDataFrame(data, columns)

my_udf = udf(lambda s: len(s))
df.withColumn('Len Value' ,my_udf(col('Value')) ).show()
Copy

출력

Snowpark에는 Udf 함수에 대한 입력 및 반환 유형이 필요합니다. 공급자가 제공하지 않고 SMA 는 매개 변수를 사용할 수 없기 때문입니다.

from snowflake.snowpark import Session, DataFrameStatFunctions
from snowflake.snowpark.functions import col, udf

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 'Test 1'],
        ['Q2', 'Test 2'],
        ['Q3', 'Test 1'],
        ['Q4', 'Test 1']]

columns = ['Quadrant', 'Value']
df = spark.createDataFrame(data, columns)
#EWI: SPRKPY1073 => pyspark.sql.functions.udf function without the return type parameter is not supported. See documentation for more info.
my_udf = udf(lambda s: len(s))

df.withColumn('Len Value' ,my_udf(col('Value')) ).show()
Copy

권장 수정

이 시나리오를 수정하려면 입력 및 출력의 반환 유형에 대한 가져오기를 추가한 다음 udf 함수 _my_udf에 반환 유형 및 input_types[] 의 매개 변수를 추가해야 합니다.

from snowflake.snowpark import Session, DataFrameStatFunctions
from snowflake.snowpark.functions import col, udf
from snowflake.snowpark.types import IntegerType, StringType

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 'Test 1'],
        ['Q2', 'Test 2'],
        ['Q3', 'Test 1'],
        ['Q4', 'Test 1']]

columns = ['Quadrant', 'Value']
df = spark.createDataFrame(data, columns)

my_udf = udf(lambda s: len(s), return_type=IntegerType(), input_types=[StringType()])

df.with_column("result", my_udf(df.Value)).show()
Copy
시나리오 2

PySpark 에서 매개 변수 없이 @udf 데코레이터를 사용할 수 있습니다

입력

from pyspark.sql.functions import col, udf

spark = SparkSession.builder.getOrCreate()
data = [['Q1', 'Test 1'],
        ['Q2', 'Test 2'],
        ['Q3', 'Test 1'],
        ['Q4', 'Test 1']]

columns = ['Quadrant', 'Value']
df = spark.createDataFrame(data, columns)

@udf()
def my_udf(str):
    return len(str)


df.withColumn('Len Value' ,my_udf(col('Value')) ).show()
Copy

출력

Snowpark에서는 udf 데코레이터의 모든 매개 변수가 필요합니다.

from snowflake.snowpark.functions import col, udf

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 'Test 1'],
        ['Q2', 'Test 2'],
        ['Q3', 'Test 1'],
        ['Q4', 'Test 1']]

columns = ['Quadrant', 'Value']
df = spark.createDataFrame(data, columns)

#EWI: SPRKPY1073 => pyspark.sql.functions.udf decorator without parameters is not supported. See documentation for more info.

@udf()
def my_udf(str):
    return len(str)

df.withColumn('Len Value' ,my_udf(col('Value')) ).show()
Copy

권장 수정

이 시나리오를 수정하려면 입력 및 출력의 반환 유형에 대한 가져오기를 추가한 다음 udf @udf 데코레이터에서 반환 유형 및 input_types[]의 매개 변수를 추가해야 합니다.

from snowflake.snowpark.functions import col, udf
from snowflake.snowpark.types import IntegerType, StringType

spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Q1', 'Test 1'],
        ['Q2', 'Test 2'],
        ['Q3', 'Test 1'],
        ['Q4', 'Test 1']]

columns = ['Quadrant', 'Value']
df = spark.createDataFrame(data, columns)

@udf(return_type=IntegerType(), input_types=[StringType()])
def my_udf(str):
    return len(str)

df.withColumn('Len Value' ,my_udf(col('Value')) ).show()
Copy

추가 권장 사항

SPRKPY1022

메시지: pyspark.sql.functions.log10에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.functions.log10 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 pyspark.sql.functions.log10 함수를 사용하여 EWI 를 생성하는 예제입니다. 이 예제에서는 log10 함수를 사용하여 value 열의 10진법 로그를 계산합니다.

df = spark.createDataFrame([(1,), (10,), (100,), (1000,), (10000,)], ["value"])
df_with_log10 = df.withColumn("log10_value", log10(df["value"]))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1022 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([(1,), (10,), (100,), (1000,), (10000,)], ["value"])
#EWI: SPRKPY1022 => pyspark.sql.functions.log10 has a workaround, see documentation for more info
df_with_log10 = df.withColumn("log10_value", log10(df["value"]))
Copy

권장 수정

해결 방법으로 리터럴 값 10 을 기본값으로 전달하여 Snowflake.snowpark.functions.log 함수를 사용할 수 있습니다.

df = spark.createDataFrame([(1,), (10,), (100,), (1000,), (10000,)], ["value"])
df_with_log10 = df.withColumn("log10_value", log(10, df["value"]))
Copy

추가 권장 사항

SPRKPY1016

경고

이 문제 코드는 Spark Conversion Core Version 0.11.7 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.functions.collect_set에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.collect_set 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

collect set을 사용하여 중복 없이 _colname 의 요소를 가져옵니다.

col = collect_set(colName)
Copy

출력

SMA 는 collect_set가 사용되는 라인에 EWI SPRKPY1016 을 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

#EWI: SPRKPY1016 => pyspark.sql.functions.collect_set has a workaround, see documentation for more info
col = collect_set(colName)
Copy

권장 수정

Array_agg 함수를 사용하고 두 번째 인자에 true 값을 추가합니다.

col = array_agg(col, True)
Copy

추가 권장 사항

SPRKPY1047

경고

이 문제 코드는 Spark Conversion Core Version 4.6.0 이후 사용 중단 되었습니다.

설명

이 문제는 도구가 pyspark.context.SparkContext.setLogLevel 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

다음은 setLogLevel 메서드의 사용 예시입니다.

sparkSession.sparkContext.setLogLevel("WARN")
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1047 을 추가합니다.

#EWI: SPRKPY1047 => pyspark.context.SparkContext.setLogLevel has a workaround, see documentation for more info
sparkSession.sparkContext.setLogLevel("WARN")
Copy

권장 수정

setLogLevel 함수 사용을 간단한 로깅 사용을 위한 편리한 함수 세트를 제공하는 logging.basicConfig 로 바꿉니다. 이를 사용하려면 “로깅”과 “시스템”이라는 두 모듈을 가져와야 하며, “레벨 등가 테이블”을 사용하여 레벨 상수를 대체해야 합니다.

import logging
import sys
logging.basicConfig(stream=sys.stdout, level=logging.WARNING)
Copy

  • 레벨 등가 테이블

레벨 소스 매개 변수

레벨 대상 매개 변수

“ALL”

해당 항목이 없습니다 _

“DEBUG”

logging.DEBUG

“ERROR”

logging.ERROR

“FATAL”

logging.CRITICAL

“INFO”

logging.INFO

“OFF”

logging.NOTSET

“TRACE”

해당 항목이 없습니다 _

“WARN”

logging.WARNING

추가 권장 사항

SPRKPY1057

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

메시지: PySpark 데이터 프레임 옵션 인자에 리터럴이 아닌 값이 포함되어 있으므로 평가할 수 없습니다

카테고리: 경고.

설명

이 문제 코드는 더 이상 사용되지 않습니다. 이전 버전을 사용 중인 경우 최신 버전으로 업그레이드하십시오.

추가 권장 사항

SPRKPY1006

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

메시지 : pyspark.context.SparkContext 는 필수가 아닙니다

카테고리 : 경고.

설명

이 문제는 도구가 Snowflake에서 필수가 아닌 pyspark.context.SparkContext 의 사용을 감지할 때 발생합니다.

시나리오

입력

이 예제에서는 Spark 클러스터에 대한 연결을 생성하는 두 가지 컨텍스트가 있습니다

from pyspark import SparkContext

sql_context1 = SparkContext(my_sc1)
sql_context2 = SparkContext(sparkContext=my_sc2)
Copy

출력

Snowflake에는 클러스터링이 없어 컨텍스트가 필요하지 않으므로 Spark 속성을 포함하는 my_sc1 및 my_sc2 변수는 필요하지 않거나 코드를 수정하기 위해 조정해야 할 수 있습니다.

from snowflake.snowpark import Session
#EWI: SPRKPY1006 => pyspark.sql.context.SparkContext is not required
sql_context1 = my_sc1
#EWI: SPRKPY1006 => pyspark.sql.context.SparkContext is not required

sql_context2 = my_sc2
Copy

추가 권장 사항

  • 불필요한 매개 변수가 제거되고 경고 설명이 삽입됩니다. 사용자의 작업이 없어야 합니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1032

메시지: _ Spark 요소 _가 정의되지 않았습니다

카테고리: 변환 오류

설명

이 문제는 SMA 에서 지정된 요소에 대한 적절한 매핑 상태를 확인할 수 없을 때 표시됩니다. 즉, SMA 는 이 요소가 Snowpark에서 지원되는지 여부를 아직 알지 못합니다. 이 코드는 정의되지 않은 요소에 대해 SMA 가 사용하는 일반적인 오류 코드입니다.

시나리오

입력

다음은 SMA 가 적절한 매핑 상태를 확인할 수 없는 함수의 예입니다. 이 경우 not_defined_function() 이 유효한 PySpark 함수이고 코드가 실행된다고 가정해야 합니다.

sc.parallelize(["a", "b", "c", "d", "e"], 3).not_defined_function().collect()
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1032 를 추가하여 이 요소가 정의되지 않았음을 알립니다.

#EWI: SPRKPY1032 => pyspark.rdd.RDD.not_defined_function is not defined
sc.parallelize(["a", "b", "c", "d", "e"], 3).not_defined_function().collect()
Copy

권장 수정

문제를 식별하기 위해 다음 유효성 검사를 수행할 수 있습니다.

  • 소스 코드의 구문이 올바른지, 철자가 올바른지 확인하십시오.

  • SMA 가 지원하는 PySpark 버전을 사용하고 있는지 확인하십시오. SMA 를 실행하는 현재 SMA 에서 지원되는 PySpark 버전을 확인하려면 DetailedReport.docx 파일의 첫 페이지를 검토하십시오.

유효한 PySpark 요소인 경우 SMA 의 문제 신고 옵션을 사용하여 해당 요소에서 변환 오류가 발생했다고 보고하고, 도움이 될 만한 추가 정보를 함께 포함해 주시기 바랍니다.

요소가 정의되어 있지 않다고 해서 Snowpark에서 지원하지 않는다는 의미는 아닙니다. Snowpark 설명서 에서 동등한 요소가 있는지 확인해야 합니다.

추가 권장 사항

SPRKPY1063

메시지: pyspark.sql.pandas.functions.pandas_udf에 해결 방법이 있습니다.

카테고리: 경고

설명

이 문제는 도구가 pyspark.sql.pandas.functions.pandas_udf 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

Pandas_udf 함수는 대량의 데이터로 작동하는 사용자 정의 함수를 생성하는 데 사용됩니다.

@pandas_udf(schema, PandasUDFType.GROUPED_MAP)
def modify_df(pdf):
    return pd.DataFrame({'result': pdf['col1'] + pdf['col2'] + 1})
df = spark.createDataFrame([(1, 2), (3, 4), (1, 1)], ["col1", "col2"])
new_df = df.groupby().apply(modify_df)
Copy

출력

SMA 에는 EWI 메시지가 추가되어 pandas_udf에 해결 방법이 있음을 나타냅니다.

#EWI: SPRKPY1062 => pyspark.sql.pandas.functions.pandas_udf has a workaround, see documentation for more info
@pandas_udf(schema, PandasUDFType.GROUPED_MAP)

def modify_df(pdf):
    return pd.DataFrame({'result': pdf['col1'] + pdf['col2'] + 1})

df = spark.createDataFrame([(1, 2), (3, 4), (1, 1)], ["col1", "col2"])

new_df = df.groupby().apply(modify_df)
Copy

권장 수정

매개 변수 유형을 새 매개 변수 input_types 으로 명시적으로 지정하고, 해당되는 경우 functionType 매개 변수를 제거합니다. 생성된 함수는 선택 문 내에서 호출해야 합니다.

@pandas_udf(
    return_type = schema,
    input_types = [PandasDataFrameType([IntegerType(), IntegerType()])]
)

def modify_df(pdf):
    return pd.DataFrame({'result': pdf['col1'] + pdf['col2'] + 1})

df = spark.createDataFrame([(1, 2), (3, 4), (1, 1)], ["col1", "col2"])

new_df = df.groupby().apply(modify_df) # You must modify function call to be a select and not an apply
Copy

추가 권장 사항

SPRKPY1078

메시지: SparkContext.setLogLevel 함수의 인자가 리터럴 값이 아니므로 평가할 수 없습니다

카테고리: 경고

설명

이 문제는 SMA 가 리터럴 값이 아닌 인자(예: 변수)와 함께 pyspark.context.SparkContext.setLogLevel 함수가 사용된 것을 감지할 때 발생합니다.

SMA 는 소스 코드의 정적 분석을 수행하므로 해당 인자의 내용을 평가하여 Snowpark에서 동등한 것을 결정할 수 없습니다.

시나리오

입력

이 예제에서는 로그 레벨을 my_log_level 변수에 정의한 다음 setLogLevel 메서드에서 my_log_level을 매개 변수로 사용합니다.

my_log_level = "WARN"
sparkSession.sparkContext.setLogLevel(my_log_level)
Copy

출력

SMA 가 로그 수준 매개 변수의 인자를 평가할 수 없으므로 변환된 로깅 라인에 EWI SPRKPY1078 이 추가됩니다.

my_log_level = "WARN"
#EWI: SPRKPY1078 => my_log_level is not a literal value and therefore could not be evaluated. Make sure the value of my_log_level is a valid level in Snowpark. Valid log levels are: logging.CRITICAL, logging.DEBUG, logging.ERROR, logging.INFO, logging.NOTSET, logging.WARNING
logging.basicConfig(stream = sys.stdout, level = my_log_level)
Copy

권장 수정

SMA 가 인자를 평가할 수 없더라도 pyspark.context.SparkContext.setLogLevel 함수를 Snowpark에 해당하는 함수로 변환합니다. 생성된 출력 코드의 수준 인자 값이 아래 테이블에 따라 Snowpark에서 유효하고 동등한 로그 수준인지 확인하십시오.

PySpark 로그 수준

Snowpark 로그 수준 동등 기능

ALL

logging.NOTSET

DEBUG

logging.DEBUG

ERROR

logging.ERROR

FATAL

logging.CRITICAL

INFO

logging.INFO

OFF

logging.WARNING

TRACE

logging.NOTSET

WARN

logging.WARNING

따라서 권장 수정 사항은 다음과 같습니다.

my_log_level = logging.WARNING
logging.basicConfig(stream = sys.stdout, level = my_log_level)
Copy

추가 권장 사항

SPRKPY1029

메시지: 이 문제는 도구가 pyspark.sql.readwriter.DataFrameReader.parquet의 사용을 감지할 때 표시됩니다. 이 함수는 지원되지만, Snowpark와 Spark API 간의 일부 차이점 때문에 일부 수동 변경이 필요할 수 있습니다.

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.readwriter.DataFrameReader.parquet 함수의 사용을 감지할 때 발생합니다. 이 함수는 Snowpark에서 지원되지만 일부 수동 변경이 필요한 몇 가지 차이점이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.readwriter.DataFrameReader.parquet 함수의 사용 예제입니다.

file_paths = [
  "path/to/your/file1.parquet",
  "path/to/your/file2.parquet",
  "path/to/your/file3.parquet",
]

df = session.read.parquet(
  *file_paths,
  mergeSchema="true",
  pathGlobFilter="*file*",
  recursiveFileLookup="true",
  modifiedBefore="2024-12-31T00:00:00",
  modifiedAfter="2023-12-31T00:00:00"
)
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1029 를 추가하여 이 함수가 Snowpark에서 지원된다는 것을 알려주지만, 일부 수동 조정이 필요합니다. Snowpark에서 지원하는 옵션은 option 함수 호출로 변환되며, 지원되지 않는 옵션은 제거됩니다. 이에 대해서는 다음 섹션에서 자세히 설명합니다.

file_paths = [
  "path/to/your/file1.parquet",
  "path/to/your/file2.parquet",
  "path/to/your/file3.parquet"
]

#EWI: SPRKPY1076 => Some of the included parameters are not supported in the parquet function, the supported ones will be added into a option method.
#EWI: SPRKPY1029 => This issue appears when the tool detects the usage of pyspark.sql.readwriter.DataFrameReader.parquet. This function is supported, but some of the differences between Snowpark and the Spark API might require making some manual changes.
df = session.read.option("PATTERN", "*file*").parquet(
  *file_paths
)
Copy

권장 수정

이 섹션에서는 paths 및 options 매개 변수를 구성하여 Snowpark에서 작동하도록 하는 방법에 대해 설명합니다.

1. paths 매개 변수

Spark에서 이 매개 변수는 로컬 또는 클라우드 위치일 수 있습니다. Snowpark는 Snowflake 스테이지 를 사용하는 클라우드 위치만 허용합니다. 따라서 임시 스테이지를 생성하고 file:// 이라는 접두사를 사용하여 각 파일을 추가할 수 있습니다.

2. options 매개 변수

Snowpark에서는 parquet 함수의 매개 변수로 다른 options 을 정의할 수 없습니다. 이 문제를 해결하려면 option 또는 options 함수를 사용하여 해당 매개 변수를 DataFrameReader 의 추가 옵션으로 지정할 수 있습니다.

Snowpark options 은 PySpark options 와 정확하게 동일하지 않으므로 일부 수동 변경이 필요할 수 있습니다. 다음은 Snowpark에서 가장 일반적인 PySpark 옵션을 구성하는 방법에 대한 자세한 설명입니다.

2.1 mergeSchema 옵션

Parquet은 스키마 진화를 지원하므로 사용자는 간단한 스키마로 시작하여 필요에 따라 점차적으로 더 많은 열을 추가할 수 있습니다. 이렇게 하면 서로 다르지만 호환되는 스키마를 가진 여러 개의 Parquet 파일이 생성될 수 있습니다. Snowflake에는 infer_schema 기능이 있어 이 작업을 수행할 필요가 없으므로 mergeSchema 옵션을 제거하기만 하면 됩니다.

2.2 pathGlobFilter 옵션

스테이지에서 파일의 하위 집합만 로딩하려면 패턴 옵션을 사용하여 로딩하려는 파일과 일치하는 정규식을 지정할 수 있습니다. 이 시나리오의 출력에서 볼 수 있듯이 SMA 는 이미 이 작업을 자동화하고 있습니다.

2.3 recursiveFileLookupstr 옵션

이 옵션은 Snowpark에서 지원되지 않습니다. 가장 좋은 방법은 pathGlobFilter 옵션과 같은 정규식을 사용하여 비슷한 결과를 얻는 것입니다.

2.4 modifiedBefore / modifiedAfter 옵션

metadata 열을 사용하여 Snowflake에서도 동일한 결과를 얻을 수 있습니다.

참고

다음 옵션은 Snowpark에서 지원되지 않음:

  • 압축

  • datetimeRebaseMode

  • int96RebaseMode

  • mergeSchema

아래는 입력 코드가 Snowpark에서 작동하도록 하기 위해 어떻게 변환해야 하는지에 대한 전체 예제입니다.

from snowflake.snowpark.column import METADATA_FILE_LAST_MODIFIED, METADATA_FILENAME

temp_stage = f'{session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {temp_stage}')

session.file.put(f"file:///path/to/your/file1.parquet", f"@{temp_stage}")
session.file.put(f"file:///path/to/your/file2.parquet", f"@{temp_stage}")
session.file.put(f"file:///path/to/your/file3.parquet", f"@{temp_stage}")

df = session.read \
  .option("PATTERN", ".*file.*") \
  .with_metadata(METADATA_FILENAME, METADATA_FILE_LAST_MODIFIED) \
  .parquet(temp_stage) \
  .where(METADATA_FILE_LAST_MODIFIED < '2024-12-31T00:00:00') \
  .where(METADATA_FILE_LAST_MODIFIED > '2023-12-31T00:00:00')
Copy

추가 권장 사항

  • Snowflake에서는 다음과 같은 다른 접근법을 활용하여 Parquet 데이터를 수집할 수 있습니다.

  • 마이그레이션을 수행할 때는 SMA 보고서를 활용하여 파일 인벤토리를 작성하고 현대화 후 데이터를 어느 스테이지/테이블에 매핑할지 결정하는 것이 좋습니다.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1039

경고

이 문제 코드는 사용 중단 되었습니다

메시지: pyspark.sql.column.Column.getItem에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.column.Column.getItem 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 pyspark.sql.column.Column.getItem 함수를 사용하여 EWI 를 생성하는 예제입니다. 이 예제에서는 getItem 함수를 사용하여 위치 및 키별로 항목을 가져옵니다.

df = spark.createDataFrame([(1, ["apple", "banana", "orange"]), (2, ["carrot", "avocado", "banana"])], ["id", "fruits"])
df.withColumn("first_fruit", col("fruits").getItem(0))

df = spark.createDataFrame([(1, {"apple": 10, "banana": 20}), (2, {"carrot": 15, "grape": 25}), (3, {"pear": 30, "apple": 35})], ["id", "fruit_quantities"])
df.withColumn("apple_quantity", col("fruit_quantities").getItem("apple"))
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1039 를 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

df = spark.createDataFrame([(1, ["apple", "banana", "orange"]), (2, ["carrot", "avocado", "banana"])], ["id", "fruits"])
#EWI: SPRKPY1039 => pyspark.sql.column.Column.getItem has a workaround, see documentation for more info
df.withColumn("first_fruit", col("fruits").getItem(0))

df = spark.createDataFrame([(1, {"apple": 10, "banana": 20}), (2, {"carrot": 15, "grape": 25}), (3, {"pear": 30, "apple": 35})], ["id", "fruit_quantities"])
#EWI: SPRKPY1039 => pyspark.sql.column.Column.getItem has a workaround, see documentation for more info
df.withColumn("apple_quantity", col("fruit_quantities").getItem("apple"))
Copy

권장 수정

해결 방법으로 필드의 이름 또는 위치를 인덱스로 사용하여 Snowpark 열 인덱서 연산자 를 사용할 수 있습니다.

df = spark.createDataFrame([(1, ["apple", "banana", "orange"]), (2, ["carrot", "avocado", "banana"])], ["id", "fruits"])
df.withColumn("first_fruit", col("fruits")[0])

df = spark.createDataFrame([(1, {"apple": 10, "banana": 20}), (2, {"carrot": 15, "grape": 25}), (3, {"pear": 30, "apple": 35})], ["id", "fruit_quantities"])
df.withColumn("apple_quantity", col("fruit_quantities")["apple"])
Copy

추가 권장 사항

SPRKPY1068

메시지: toPandas에 지원되지 않는 ArrayType 형식의 열이 포함되어 있으며 해결 방법이 있습니다.

카테고리: 경고

설명

ArrayType 유형의 열이 있는 경우 DataFrame.toPandas 가 올바르게 작동하지 않습니다. 이러한 경우의 해결 방법은 json.loads 메서드를 사용하여 해당 열을 Python Dictionary로 변환하는 것입니다.

시나리오

입력

ToPandas 는 원본 DataFrame 의 데이터를 Pandas DataFrame 으로 반환합니다.

sparkDF = spark.createDataFrame([
Row(a=1, b=2., c='string1', d=date(2000, 1, 1), e=datetime(2000, 1, 1, 12, 0)),
Row(a=2, b=3., c='string2', d=date(2000, 2, 1), e=datetime(2000, 1, 2, 12, 0))
])

pandasDF = sparkDF.toPandas()
Copy

출력

이 도구는 ArrayType 유형의 열이 있는 경우 toPandas가 지원되지 않는다는 것을 알리기 위해 EWI 를 추가하지만 해결 방법이 있습니다.

sparkDF = spark.createDataFrame([
Row(a=1, b=2., c='string1', d=date(2000, 1, 1), e=datetime(2000, 1, 1, 12, 0)),
Row(a=2, b=3., c='string2', d=date(2000, 2, 1), e=datetime(2000, 1, 2, 12, 0))
])
#EWI: SPRKPY1068 => toPandas doesn't work properly If there are columns of type ArrayType. The workaround for these cases is converting those columns into a Python Dictionary by using json.loads method. example: df[colName] = json.loads(df[colName]).
pandasDF = sparkDF.toPandas()
Copy

권장 수정

pandas_df = sparkDF.toPandas()​
​
## check/convert all resulting fields from calling toPandas when they are of
## type ArrayType,
## they will be reasigned by converting them into a Python Dictionary
## using json.loads method​
​
for field in pandas_df.schema.fields:
    if isinstance(field.datatype, ArrayType):
        pandas_df[field.name] = pandas_df[field.name].apply(lambda x: json.loads(x) if x is not None else x)
Copy

추가 권장 사항

SPRKPY1048

경고

이 문제 코드는 Spark Conversion Core Version 2.4.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.session.SparkSession.conf에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.session.SparkSession.conf 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

다음은 conf 속성에 구성을 설정하는 방법의 예입니다.

spark.conf.set("spark.sql.crossJoin.enabled", "true")
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1048 을 추가합니다.

#EWI: SPRKPY1048 => pyspark.sql.session.SparkSession.conf has a workaround, see documentation for more info
spark.conf.set("spark.sql.crossJoin.enabled", "true")
Copy

권장 수정

SparkSession.conf는 Pyspark에서만 사용하는 일부 특정 설정을 전달하는 데 사용되며 Snowpark에는 적용되지 않습니다. 코드를 제거하거나 설명할 수 있습니다

#spark.conf.set("spark.sql.crossJoin.enabled", "true")
Copy

추가 권장 사항

SPRKPY1019

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.functions.datediff에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.datediff 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제에서는 datediff를 사용하여’오늘’과 다른 날짜의 요일 차이를 계산합니다.

contacts = (contacts
            #days since last event
            .withColumn('daysSinceLastEvent', datediff(lit(today),'lastEvent'))
            #days since deployment
            .withColumn('daysSinceLastDeployment', datediff(lit(today),'lastDeploymentEnd'))
            #days since online training
            .withColumn('daysSinceLastTraining', datediff(lit(today),'lastTraining'))
            #days since last RC login
            .withColumn('daysSinceLastRollCallLogin', datediff(lit(today),'adx_identity_lastsuccessfullogin'))
            #days since last EMS login
            .withColumn('daysSinceLastEMSLogin', datediff(lit(today),'vms_lastuserlogin'))
           )
Copy

출력

SMA 는 datediff가 사용되는 라인 위에 EWI SPRKPY1019 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

from pyspark.sql.functions import datediff
#EWI: SPRKPY1019 => pyspark.sql.functions.datediff has a workaround, see documentation for more info
contacts = (contacts
            #days since last event
            .withColumn('daysSinceLastEvent', datediff(lit(today),'lastEvent'))
            #days since deployment
            .withColumn('daysSinceLastDeployment', datediff(lit(today),'lastDeploymentEnd'))
            #days since online training
            .withColumn('daysSinceLastTraining', datediff(lit(today),'lastTraining'))
            #days since last RC login
            .withColumn('daysSinceLastRollCallLogin', datediff(lit(today),'adx_identity_lastsuccessfullogin'))
            #days since last EMS login
            .withColumn('daysSinceLastEMSLogin', datediff(lit(today),'vms_lastuserlogin'))
           )
Copy

SMA 는 pyspark.sql.functions.datediff를 두 날짜 사이의 일수 차이를 계산하는 Snowflake.snowpark.functions.daydiff 로 변환합니다.

권장 수정

datediff(부분: string, 종료: ColumnOrName, 시작: ColumnOrName)

작업: 날짜 시간 부분 에 대한 추가 매개 변수가 필요하고 날짜 간의 차이를 계산하는 데 더 다양한 함수를 제공하는 datediff 함수에 대한 구현이 포함된 Snowflake.snowpark.functions를 가져옵니다.

from snowflake.snowpark import Session
from snowflake.snowpark.functions import datediff
contacts = (contacts
            #days since last event
            .withColumn('daysSinceLastEvent', datediff('day', lit(today),'lastEvent'))
            #days since deployment
            .withColumn('daysSinceLastDeployment', datediff('day',lit(today),'lastDeploymentEnd'))
            #days since online training
            .withColumn('daysSinceLastTraining', datediff('day', lit(today),'lastTraining'))
            #days since last RC login
            .withColumn('daysSinceLastRollCallLogin', datediff('day', lit(today),'adx_identity_lastsuccessfullogin'))
            #days since last EMS login
            .withColumn('daysSinceLastEMSLogin', datediff('day', lit(today),'vms_lastuserlogin'))
           )
Copy

권장 사항

SPRKPY1009

메시지 : pyspark.sql.dataframe.DataFrame.approxQuantile에 해결 방법이 있습니다

카테고리 : 경고.

설명

이 문제는 도구가 pyspark.sql.dataframe.DataFrame.approxQuantile 의 사용을 감지했을 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

Pyspark는 두 가지 다른 approxQuantile 함수를 사용한다는 것을 이해하는 것이 중요합니다. 여기서는 DataFrame approxQuantile 버전을 사용합니다.

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
data = [['Sun', 10],
        ['Mon', 64],
        ['Thr', 12],
        ['Wen', 15],
        ['Thu', 68],
        ['Fri', 14],
        ['Sat', 13]]

columns = ['Day', 'Ammount']
df = spark.createDataFrame(data, columns)
df.approxQuantile('Ammount', [0.25, 0.5, 0.75], 0)
Copy

출력

SMA 는 접근 사분위수가 사용되는 라인 위에 EWI SPRKPY1009 을 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

from snowflake.snowpark import Session
spark = Session.builder.getOrCreate()
spark.update_query_tag({"origin":"sf_sit","name":"sma","version":{"major":0,"minor":0,"patch":0},"attributes":{"language":"Python"}})
data = [['Sun', 10],
        ['Mon', 64],
        ['Thr', 12],
        ['Wen', 15],
        ['Thu', 68],
        ['Fri', 14],
        ['Sat', 13]]

columns = ['Day', 'Ammount']
df = spark.createDataFrame(data, columns)
#EWI: SPRKPY1009 => pyspark.sql.dataframe.DataFrame.approxQuantile has a workaround, see documentation for more info
df.approxQuantile('Ammount', [0.25, 0.5, 0.75], 0)
Copy

권장 수정

Snowpark approxQuantile 메서드를 사용합니다. 일부 매개 변수가 일치하지 않으므로 일부 수동 조정이 필요합니다. 출력 코드 예제의 경우 권장 수정 사항이 될 수 있습니다.

from snowflake.snowpark import Session
...
df = spark.createDataFrame(data, columns)

df.stat.approx_quantile('Ammount', [0.25, 0.5, 0.75])
Copy

SnowPark 에는 DataFrame.approxQuantile의 상대 오류 매개 변수가 존재하지 않습니다.

추가 권장 사항

SPRKPY1058

메시지: < 방법 > < 키 > 플랫폼별 키는 지원되지 않습니다.

카테고리: ConversionError

설명

pyspark.sql.conf.RuntimeConfig 의 get 및 set 메서드는 플랫폼별 키로 지원되지 않습니다.

시나리오

get 또는 set 메서드를 사용하는 모든 경우 출력 코드에 EWI 가 포함되는 것은 아닙니다. 이 EWI 는 도구가 지원되지 않는 플랫폼별 키로 이러한 메서드의 사용을 감지하면 표시됩니다.

시나리오 1

입력

다음은 Snowpark에서 지원되는 키를 사용한 get 또는 set 메서드의 예입니다.

session.conf.set("use_constant_subquery_alias", False)
spark.conf.set("sql_simplifier_enabled", True)

session.conf.get("use_constant_subquery_alias")
session.conf.get("use_constant_subquery_alias")
Copy

출력

Snowpark에서 키가 지원되므로 이 도구는 출력 코드에 EWI 를 추가하지 않습니다.

session.conf.set("use_constant_subquery_alias", True)
session.conf.set("sql_simplifier_enabled", False)

session.conf.get("use_constant_subquery_alias")
session.conf.get("sql_simplifier_enabled")
Copy

권장 수정

이 시나리오에 대한 권장 수정 사항은 없습니다.

시나리오 2

입력

아래는 지원되지 않는 키를 사용한 예제입니다.

data =
    [
      ("John", 30, "New York"),
      ("Jane", 25, "San Francisco")
    ]

session.conf.set("spark.sql.shuffle.partitions", "50")
spark.conf.set("spark.yarn.am.memory", "1g")

session.conf.get("spark.sql.shuffle.partitions")
session = spark.conf.get("spark.yarn.am.memory")

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])
Copy

출력

도구는 출력 코드에 EWI SPRKPY1058 을 추가하여 이러한 메서드가 플랫폼별 키에서 지원되지 않음을 알려줍니다.

data =
    [
      ("John", 30, "New York"),
      ("Jane", 25, "San Francisco")
    ]

#EWI: SPRKPY1058 => pyspark.sql.conf.RuntimeConfig.set method with this "spark.sql.shuffle.partitions" Platform specific key is not supported.
spark.conf.set("spark.sql.shuffle.partitions", "50")
#EWI: SPRKPY1058 => pyspark.sql.conf.RuntimeConfig.set method with this "spark.yarn.am.memory" Platform specific key is not supported.
spark.conf.set("spark.yarn.am.memory", "1g")

#EWI: SPRKPY1058 => pyspark.sql.conf.RuntimeConfig.get method with this "spark.sql.shuffle.partitions" Platform specific key is not supported.
spark.conf.get("spark.sql.shuffle.partitions")
#EWI: SPRKPY1058 => pyspark.sql.conf.RuntimeConfig.get method with this "spark.yarn.am.memory" Platform specific key is not supported.
spark.conf.get("spark.yarn.am.memory")

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])
Copy

권장 수정

권장되는 수정 방법은 이러한 메서드를 제거하는 것입니다.

data =
    [
      ("John", 30, "New York"),
      ("Jane", 25, "San Francisco")
    ]

df = spark.createDataFrame(data, schema=["Name", "Age", "City"])
Copy

추가 권장 사항

SPRKPY1049

경고

이 문제 코드는 Spark Conversion Core Version 2.1.9 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.session.SparkSession.sparkContext에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.session.SparkSession.sparkContext 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

아래는 Spark 세션을 생성한 다음 SparkContext 속성을 사용하여 앱 이름을 인쇄하는 예제입니다.

print("APP Name :"+spark.sparkContext.appName())
Copy

출력

이 도구는 해결 방법을 구현할 수 있음을 나타내는 EWI SPRKPY1049 를 추가합니다.

#EWI: SPRKPY1049 => pyspark.sql.session.SparkSession.sparkContext has a workaround, see documentation for more info
print("APP Name :"+spark.sparkContext.appName())
Copy

권장 수정

SparkContext 는 SnowPark 에서 지원되지 않지만, SparkContext 의 세션 인스턴스에서 직접 메서드와 속성에 액세스할 수 있습니다.

## Pyspark
print("APP Name :"+spark.sparkContext.appName())
can be used in SnowPark removing the sparkContext as:
#Manual adjustment in SnowPark
print("APP Name :"+spark.appName());
Copy

추가 권장 사항

SPRKPY1018

경고

이 문제 코드는 Spark Conversion Core Version 4.8.0 이후에 사용 중단 되었습니다.

메시지: pyspark.sql.functions.date_sub에 해결 방법이 있습니다

카테고리: 경고.

설명

이 문제는 도구가 pyspark.sql.functions.date_sub 의 사용을 감지할 때 발생하며 해결이 가능합니다.

시나리오

입력

이 예제에서는 date_add를 사용하여 데이터프레임 df의 현재 날짜 5일 전 날짜를 계산합니다.

col = df.select(date_sub(df.colName, 5))
Copy

출력

SMA 는 date_sub가 사용되는 라인 위에 EWI SPRKPY1018 를 반환하므로 수정할 위치를 식별하는 데 사용할 수 있습니다.

#EWI: SPRKPY1018 => pyspark.sql.functions.date_sub has a workaround, see documentation for more info
col = df.select(date_sub(df.colName, 5))
Copy

권장 수정

Date_sub 함수에 대한 구현이 포함된 Snowflake.snowpark.functions를 가져옵니다.

from pyspark.sql.functions import date_sub
df.withColumn("date", date_sub(df.colName, 5))
Copy

추가 권장 사항

SPRKPY1008

메시지: pyspark.sql.context.HiveContext 는 필수가 아닙니다

카테고리: 경고.

설명

이 문제는 도구가 필수가 아닌 pyspark.sql.context.HiveContext 의 사용을 감지할 때 발생합니다.

시나리오

입력

이 예제에서는 Hive 스토어에 대한 연결을 생성하는 예제입니다.

from pyspark.sql import HiveContext
hive_context = HiveContext(sc)
df = hive_context.table("myTable")
df.show()
Copy

출력

Snowflake에는 Hive 스토어가 없으므로 Hive 컨텍스트가 필요하지 않으며, 방법을 알아보려면 이 자습서 를 확인하여 Snowflake에서 Parquet 파일을 계속 사용할 수 있습니다.

#EWI: SPRKPY1008 => pyspark.sql.context.HiveContext is not required
hive_context = sc
df = hive_context.table("myTable")
df.show()
Copy

sc 변수는 Snow Park Session Object 를 참조합니다.

권장 수정

예제의 출력 코드의 경우 이 코드와 유사한 Snow Park 세션 오브젝트 를 추가해야 합니다.

## Here manually we can add the Snowpark Session object via a json config file called connection.json
import json
from snowflake.snowpark import Session
jsonFile = open("connection.json")
connection_parameter = json.load(jsonFile)
jsonFile.close()
sc = Session.builder.configs(connection_parameter).getOrCreate()

hive_context = sc
df = hive_context.table("myTable")
df.show()
Copy

추가 권장 사항

SPRKPY1059

경고

이 문제 코드는 Spark Conversion Core Version 2.45.1 이후에 사용 중단 되었습니다.

메시지: pyspark.storagelevel.StorageLevel 에 해결 방법이 있습니다. 설명서를 참조하십시오.

카테고리: 경고

설명

현재 Snowpark에서는 Snowflake가 저장소를 제어 하므로 StorageLevel 사용은 필수가 아닙니다. 자세한 내용은 EWI SPRKPY1072에서 확인할 수 있습니다

추가 권장 사항

  • 애플리케이션을 최신 버전으로 업그레이드하십시오.

  • 추가 지원이 필요하면 sma-support@Snowflake.com 또는 SMA에 문제를 게시할 수 있습니다.

SPRKPY1079

메시지: SparkContext.setLogLevel 함수의 인자가 유효한 PySpark 로그 수준이 아닙니다

카테고리: 경고

설명

이 문제는 SMA 가 PySpark 에서 유효한 로그 레벨이 아닌 인자로 pyspark.context.SparkContext.setLogLevel 함수가 사용된 것을 감지하여 Snowpark에서 이에 상응하는 함수를 찾을 수 없기 때문에 발생합니다.

시나리오

입력

여기서 로그 수준은 유효한 Pyspark 로그 수준이 아닌 “INVALID_LOG_LEVEL”을 사용합니다.

sparkSession.sparkContext.setLogLevel("INVALID_LOG_LEVEL")
Copy

출력

SMA 가 로깅 수준 “INVALID_LOG_LEVEL”을 인식할 수 없는 경우, SMA 가 변환을 수행하더라도 EWI SPRKPY1079 가 추가되어 문제가 있을 수 있음을 나타냅니다.

#EWI: SPRKPY1079 => INVALID_LOG_LEVEL is not a valid PySpark log level, therefore an equivalent could not be determined in Snowpark. Valid PySpark log levels are: ALL, DEBUG, ERROR, FATAL, INFO, OFF, TRACE, WARN
logging.basicConfig(stream = sys.stdout, level = logging.INVALID_LOG_LEVEL)
Copy

권장 수정

pyspark.context.SparkContext.setLogLevel 함수에서 사용되는 로그 수준이 PySpark 또는 Snowpark 에서유효한 로그 수준인지 확인하고 다시 시도합니다.

logging.basicConfig(stream = sys.stdout, level = logging.DEBUG)
Copy

추가 권장 사항

SPRKPY1028

메시지: pyspark.sql.readwriter.DataFrameReader.orc에 해결 방법이 있습니다. 자세한 내용은 설명서를 참조하십시오

카테고리: 경고

설명

이 문제는 SMA 가 pyspark.sql.readwriter.DataFrameReader.orc 함수의 사용을 감지할 때 발생하며, 해결 방법이 있습니다.

시나리오

입력

다음은 EWI 를 생성하는 pyspark.sql.readwriter.DataFrameReader.orc 함수의 사용 예제입니다. 이 예제에서는 orc 함수를 사용하여 여러 .orc 파일을 읽고 mergeSchema 및 recursiveFileLookup 같은 추가 옵션을 사용하여 파일 읽기 동작을 미세 조정합니다.

file_paths = [
  "path/to/your/file1.orc",
  "path/to/your/file2.orc",
  "path/to/your/file3.orc",
]

df = session.read.orc(
  file_paths,
  mergeSchema="True",
  recursiveFileLookup="True"
)
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1028 을 추가하여 이 함수가 Snowpark에서 직접 지원되지는 않지만 우회 방법이 있음을 알려줍니다.

file_paths = [
  "path/to/your/file1.orc",
  "path/to/your/file2.orc",
  "path/to/your/file3.orc",
]

#EWI: SPRKPY1028 => pyspark.sql.readwriter.DataFrameReader.orc has a workaround, see documentation for more info
df = session.read.orc(
  file_paths,
  mergeSchema="True",
  recursiveFileLookup="True"
)
Copy

권장 수정

이 섹션에서는 경로 매개 변수와 추가 옵션 을 구성하여 Snowpark에서 작동하도록 하는 방법을 설명합니다.

1. path 매개 변수

Snowpark는 경로 매개 변수를 스테이지 위치로 요구하므로 해결 방법으로 임시 스테이지를 생성하고 file:// 접두사를 사용하여 각 .orc 파일을 해당 스테이지에 추가할 수 있습니다.

2. options 매개 변수

Snowpark에서는 extra options 을 orc 함수의 매개 변수로 정의하는 것을 허용하지 않습니다. 해결 방법으로 많은 경우 DataFrameReader.option 함수를 사용하여 해당 매개 변수를 DataFrameReader 의 옵션으로 지정할 수 있습니다.

참고

다음 옵션은 Snowpark에서 지원되지 않음:

  • 압축

  • mergeSchema

아래는 위에서 언급한 제안을 적용하여 Snowpark에서 작동하도록 한 후 입력 코드의 전체 예시입니다.

stage = f'{session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {stage}')

session.file.put(f"file:///path/to/your/file1.orc", f"@{stage}")
session.file.put(f"file:///path/to/your/file2.orc", f"@{stage}")
session.file.put(f"file:///path/to/your/file3.orc", f"@{stage}")

df = session.read.option(recursiveFileLookup, "True").orc(stage)
Copy

추가 권장 사항

SPRKPY1038

메시지: _ Spark 요소 _는 아직 인식되지 않습니다

카테고리: 변환 오류

설명

이 문제는 소스 코드에 SMA 가 인식하지 못한 PySpark 요소가 있을 때 표시됩니다. 이는 다음과 같은 다양한 이유로 발생할 수 있습니다.

  • PySpark 에 존재하지 않는 요소입니다.

  • SMA 버전에서 아직 지원하지 않는 PySpark 버전에 추가된 요소입니다.

  • 요소를 처리할 때 SMA 의 내부 오류입니다.

인식되지 않는 요소에 대해 SMA 에서 사용하는 일반적인 오류 코드입니다.

시나리오

입력

아래는 PySpark 에 존재하지 않아 SMA 에서 인식할 수 없는 함수를 사용한 예제입니다.

from pyspark.sql import functions as F
F.unrecognized_function()
Copy

출력

SMA 는 출력 코드에 EWI SPRKPY1038 을 추가하여 이 요소를 인식할 수 없음을 알립니다.

from snowflake.snowpark import functions as F
#EWI: SPRKPY1038 => pyspark.sql.functions.non_existent_function is not yet recognized
F.unrecognized_function()
Copy

권장 수정

문제를 식별하기 위해 다음 유효성 검사를 수행할 수 있습니다.

  • PySpark 에 요소가 있는지 확인합니다.

  • 요소의 철자가 올바른지 확인합니다.

  • SMA 가 지원하는 PySpark 버전을 사용하고 있는지 확인하십시오. SMA 를 실행하는 현재 SMA 에서 지원되는 PySpark 버전을 확인하려면 DetailedReport.docx 파일의 첫 페이지를 검토하십시오.

유효한 PySpark 요소인 경우 SMA 의 문제 신고 옵션을 사용하여 해당 요소에서 변환 오류가 발생했다고 보고하고, 도움이 될 만한 추가 정보를 함께 포함해 주시기 바랍니다.

SMA 가 요소를 인식할 수 없는 경우, 해당 요소가 Snowpark에서 지원되지 않는다는 의미는 아닙니다. Snowpark 설명서 에서 동등한 요소가 있는지 확인해야 합니다.

추가 권장 사항

SPRKPY1069

메시지: PartitionBy 매개 변수가 목록인 경우, Snowpark는 오류를 throw합니다.

카테고리: 경고

설명

DataFrameWriter.parquet 메서드에서 매개 변수 partitionBy 를 사용하는 경우 도구는 EWI 를 표시합니다.

이는 Snowpark에서 DataFrameWriter.parquet 이 ColumnOrSqlExpr 를 partitionBy 매개 변수로만 지원하기 때문입니다.

시나리오

시나리오 1

입력 코드:

이 시나리오에서 partitionBy 매개 변수는 목록이 아닙니다.

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])

df.write.parquet(file_path, partitionBy="age")
Copy

출력 코드:

이 도구는 EWI SPRKPY1069 를 추가하여 매개 변수가 목록인 경우 Snowpark가 오류를 throw한다는 것을 알려줍니다.

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])

#EWI: SPRKPY1069 => If partitionBy parameter is a list, Snowpark will throw and error.
df.write.parquet(file_path, partition_by = "age", format_type_options = dict(compression = "None"))
Copy

권장 수정

이 시나리오에 대한 권장 수정 사항은 없습니다. 이 도구는 partitionBy 매개 변수가 목록인 경우에 대비하여 항상 EWI 를 추가하기 때문입니다. Snowpark에서는 Snowflake 스테이지 를 사용하는 클라우드 위치만 허용됩니다.

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])

stage = f'{Session.get_fully_qualified_current_schema()}.{_generate_prefix("TEMP_STAGE")}'
Session.sql(f'CREATE TEMPORARY STAGE IF NOT EXISTS {stage}').show()
Session.file.put(f"file:///path/to/data/file.parquet", f"@{stage}")

df.write.parquet(stage, partition_by = "age", format_type_options = dict(compression = "None"))
Copy
시나리오 2

입력 코드:

이 시나리오에서 partitionBy 매개 변수는 목록입니다.

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])

df.write.parquet(file_path, partitionBy=["age", "name"])
Copy

출력 코드:

이 도구는 EWI SPRKPY1069 를 추가하여 매개 변수가 목록인 경우 Snowpark가 오류를 throw한다는 것을 알려줍니다.

df = spark.createDataFrame([(25, "Alice", "150"), (30, "Bob", "350")], schema=["age", "name", "value"])

#EWI: SPRKPY1069 => If partitionBy parameter is a list, Snowpark will throw and error.
df.write.parquet(file_path, partition_by = ["age", "name"], format_type_options = dict(compression = "None"))
Copy

권장 수정

매개 변수 값이 list 인 경우 ColumnOrSqlExpr 로 바꿉니다.

df.write.parquet(file_path, partition_by = sql_expr("age || name"), format_type_options = dict(compression = "None"))
Copy

추가 권장 사항

언어: 한국어