Snowpark Migration Accelerator: Problemcodes für Python¶

Szenarien¶

Es gibt einige Szenarien, je nachdem, ob die Optionen unterstützt werden oder nicht oder welches Format zum Schreiben der Datei verwendet wird.

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

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

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")

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")

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

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

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")

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")

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

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

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")

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")

Zusätzliche Empfehlungen¶

• Da es einige nicht unterstützte Parameter gibt, empfiehlt es sich, die Äquivalenz-Tabelle zu überprüfen und die Verhaltensweise nach der Transformation zu kontrollieren.

• Äquivalenzen-Tabelle:

Snowpark kann eine Liste von Äquivalenzen für einige Parameter unterstützen:

Wenn der verwendete Parameter nicht in der Liste enthalten ist, gibt API einen Fehler aus.

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenarien¶

Es gibt einige Szenarien, je nachdem, ob die Option unterstützt wird oder nicht oder welches Format zum Schreiben der Datei verwendet wird.

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

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

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")

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")

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

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

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")

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")

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

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

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")

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")

Zusätzliche Empfehlungen¶

• Da es einige nicht unterstützte Parameter gibt, empfiehlt es sich, die Äquivalenz-Tabelle zu überprüfen und die Verhaltensweise nach der Transformation zu kontrollieren.

• Äquivalenzen-Tabelle:

Wenn der verwendete Parameter nicht in der Liste enthalten ist, gibt API einen Fehler aus.

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Es ist wichtig zu verstehen, dass Pyspark zwei verschiedene approxQuantile-Funktionen verwendet. Hier verwenden wir die DataFrameStatFunctions approxQuantile-Version.

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)

Ausgabe

SMA gibt die EWI SPRKPY1011 über die Zeile zurück, in der approxQuantile verwendet wird, so dass Sie damit feststellen können, wo Sie korrigieren müssen.

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)

Empfohlene Korrektur

Sie können die approxQuantile-Methode von Snowpark verwenden. Einige Parameter stimmen nicht überein, so dass sie manuell angepasst werden müssen. Für das Beispiel des Ausgabecodes könnte eine empfohlene Korrektur lauten:

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])

pyspark.sql.dataframe.DataFrame.approxQuantile’s relativeError-Parameter existiert nicht in SnowPark.

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Im Folgenden finden Sie ein Beispiel für die Verwendung der pyspark.sql.functions.explode-Funktion, die diese EWI erzeugt. In diesem Beispiel wird die explode-Funktion verwendet, um eine Zeile pro Array-Element für die numbers-Spalte zu erzeugen.

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"))

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1040 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht direkt von Snowpark unterstützt wird, aber es eine Problemumgehung gibt.

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"))

Empfohlene Korrektur

Als Problemumgehung können Sie das Paket snowpark_extensions importieren, das eine Erweiterung für die Funktion explode bietet.

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"))

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

In Pyspark können Sie Leerzeichen und Tabulatoren für die Identifikationsebene mischen.

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

Ausgabe

SMA kann nicht mit gemischten Einrückungsmarkierungen umgehen. Wenn dies in einer Python-Code-Datei erkannt wird, fügt SMA die EWI SPRKPY1074 in der ersten Zeile hinzu.

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

Empfohlene Korrektur

Die Lösung besteht darin, alle Einrückungssymbole anzugleichen.

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

¶

Szenario¶

Eingabe

Im Folgenden finden Sie ein Beispiel für die Verwendung der Funktion pyspark.sql.functions.ntile, die diese EWI erzeugt. In diesem Beispiel wird die Funktion ntile verwendet, um die Zeilen in 3 Buckets zu unterteilen.

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

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1025 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht direkt von Snowpark unterstützt wird, aber es eine Problemumgehung gibt.

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

Empfohlene Korrektur

Snowpark hat eine äquivalente ntile-Funktion, allerdings sollte das Argument, das an diese Funktion übergeben wird, eine Spalte sein. Als Problemumgehung können Sie das Literal-Argument mit der Funktion snowflake.snowpark.functions.lit in eine Spalte umwandeln.

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

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Im Folgenden finden Sie ein Beispiel für die Verwendung der pyspark.sql.dataframe.DataFrame.writeTo-Funktion. Der Datenframe df wird in einen Tabellennamen Personal_info geschrieben.

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

df.writeTo("Personal_info")

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1087 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht unterstützt wird, aber eine Problemumgehung besteht.

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")

Empfohlene Korrektur

Die Problemumgehung besteht darin, stattdessen die Snowpark DataFrameWriter SaveAsTable-Methode zu verwenden.

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

df.write.saveAsTable("Personal_info")

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Nachfolgend finden Sie ein Beispiel für die Verwendung der pyspark.sql.functions.reverse-Funktion, die diese EWI erzeugt. In diesem Beispiel wird die reverse-Funktion verwendet, um jede Zeichenfolge in der word-Spalte umzukehren.

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

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1035 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht direkt von Snowpark unterstützt wird, aber es eine Problemumgehung gibt.

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"))

Empfohlene Korrektur

Als Problemumgehung können Sie das Paket snowpark_extensions importieren, das eine Erweiterung für die Funktion reverse bietet.

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"))

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Nachfolgend sehen Sie ein Beispiel mit einem der Elemente, die diesen EWI auslösen.

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

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1064 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht anwendbar ist.

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

Empfohlene Korrektur

Die SMA entfernt die Importanweisung und fügt die Ausgabe dem Bestand Issues.csv hinzu, wobei alle Verwendungen des Spark-Elements entfernt werden.

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

Zusätzliche Empfehlungen¶

• Sehen Sie sich die Dokumentation zu Snowpipe an, um zu sehen, wie es zum aktuellen Szenario passt.

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Nachfolgend finden Sie ein Beispiel, das eine Variable mit conf.set setzt.

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

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

Ausgabe

Das Tool fügt die Adresse EWI SPRKPY1050 hinzu und weist darauf hin, dass eine Problemumgehung implementiert werden kann.

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")

Empfohlene Korrektur

SparkConf.set wird verwendet, um eine Konfigurationseinstellung zu setzen, die nur von Pyspark verwendet wird und nicht für Snowpark gilt. Sie können den Code entfernen oder kommentieren

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

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario ¶

Eingabe: Die EWI-Meldung wird angezeigt, wenn der Code eine ungültige Syntax hat, zum Beispiel:

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

Ausgabe: SMA Finden Sie einen Parsing-Fehler und kommentieren Sie den Parsing-Fehler, indem Sie die entsprechende EWI-Meldung hinzufügen:

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

Szenario¶

Eingabe

Im Folgenden finden Sie ein Beispiel für die Verwendung der pyspark.sql.functions.last-Funktion, die diese EWI erzeugt. In diesem Beispiel wird die Funktion last verwendet, um den letzten Wert für jeden Namen zu erhalten.

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"))

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1021 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht direkt von Snowpark unterstützt wird, aber es eine Problemumgehung gibt.

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"))

Empfohlene Korrektur

Als Problemumgehung können Sie die Snowflake LAST_VALUE-Funktion verwenden. Um diese Funktion von Snowpark aus aufzurufen, verwenden Sie die Funktion snowflake.snowpark.functions.call_builtin und übergeben die last_value-Zeichenfolge als erstes Argument und die entsprechende Spalte als zweites Argument. Wenn Sie den Namen der Spalte in der last-Funktion verwendet haben, sollten Sie ihn beim Aufruf der call_builtin-Funktion in eine Spalte umwandeln.

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"))

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Beschreibung: >- Der mode-Parameter in den Methoden von CSV, JSON und PARQUET wird in overwrite umgewandelt.

Szenarien¶

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

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

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

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

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

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenarien¶

Die Spark-Signatur für diese Methode DataFrameWriter.save(path, format, mode, partitionBy, **options) existiert nicht in Snowpark. Daher wird jede Verwendung der Ladefunktion eine EWI im Ausgabecode haben.

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")

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")

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")

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")

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")

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)

Szenarien¶

Es gibt mehrere Szenarien, die von der Anzahl der an die Methode übergebenen Parameter abhängen.

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

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

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

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

In diesem Beispiel berechnet pyspark den atanh für einen Datenframe mit pyspark.sql.functions.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"]))

Ausgabe

SMA gibt die EWI SPRKPY1015 über die Zeile zurück, in der atanh verwendet wird, so dass Sie damit feststellen können, wo Sie korrigieren müssen.

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"]))

Empfohlene Korrektur

Es gibt keine direkte „atanh“-Implementierung, aber“call_function“ kann stattdessen verwendet werden, wobei „atanh“ als erster Parameter und colName als zweiter Parameter verwendet wird.

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')))

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

SparkConf kann ohne Parameter oder mit loadDefaults aufgerufen werden.

from pyspark import SparkConf

my_conf = SparkConf(loadDefaults=True)

Ausgabe

In beiden Fällen (mit oder ohne Parameter) erstellt SMA ein Snowpark Session.builder-Objekt:

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

Zusätzliche Empfehlungen¶

• Hier wird ein unnötiger Parameter entfernt und ein Warnkommentar eingefügt. Der Benutzer muss nichts weiter tun.

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenarien¶

Es gibt einige Szenarien, die von der Art des Formats abhängen, das Sie zu laden versuchen. Es kann ein unterstütztes oder nicht unterstütztes Format sein.

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

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

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

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

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

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

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

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

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')

Zusätzliche Empfehlungen¶

• Der Snowpark-Speicherort akzeptiert nur Cloud-Speicherorte, die einen Snowflake-Stagingbereich verwenden.

• Die Dokumentation der von Snowpark unterstützten Methoden finden Sie in der Dokumentation

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Da der Authentifizierungsmechanismus in Snowpark anders ist, entfernt das Tool die Verwendungen und erstellt stattdessen eine Verbindungskonfigurationsdatei (connection.json).

from pyspark import SparkConf

my_conf = SparkConf(loadDefaults=True)

Ausgabe

Das Tool fügt die EWI SPRKPY1060 hinzu, die angibt, dass es sich um einen anderen Authentifizierungsmechanismus handelt.

#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

Empfohlene Korrektur

Um eine Verbindung zu erstellen, müssen Sie die Informationen in der connection.json-Datei eingeben.

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

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Im Folgenden finden Sie ein Beispiel für die Verwendung der pyspark.sql.column.Column.contains-Funktion, die diese EWI erzeugt. In diesem Beispiel wird die Funktion contains verwendet, um die Zeilen zu filtern, in denen die Spalte „Stadt“ die Teilzeichenkette „Neu“ enthält.

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

Ausgabe

Die SMA fügt dem Ausgabecode die EWI SPRKPY1031 hinzu, um Sie darauf hinzuweisen, dass diese Funktion nicht direkt von Snowpark unterstützt wird, aber es eine Problemumgehung gibt.

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"))

Empfohlene Korrektur

Als Problemumgehung können Sie die snowflake.snowpark.functions.contains-Funktion verwenden, indem Sie die Spalte als erstes Argument und das zu suchende Element als zweites Argument übergeben. Wenn es sich bei dem zu suchenden Element um einen Literalwert handelt, sollte dieser mit der Funktion lit in einen Spaltenausdruck umgewandelt werden.

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")))

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Hier ist ein einfaches Beispiel für die Verwendung von pyspark instr:

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

Ausgabe:

SMA gibt die EWI SPRKPY1020 über die Zeile zurück, in der instr verwendet wird, so dass Sie damit feststellen können, wo Sie korrigieren müssen.

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

Empfohlene Korrektur

Erfordert eine manuelle Änderung, indem Sie die Funktion charindex verwenden und die Reihenfolge der ersten beiden Parameter ändern.

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

Zusätzliche Empfehlung¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.

Szenario¶

Eingabe

Die getNumPartitions gibt die Anzahl der Partitionen auf einer RDD zurück.

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

print(df.getNumPartitions())

Ausgabe

Das Tool fügt diese EWI hinzu, um Sie darauf hinzuweisen, dass getNumPartitions nicht erforderlich ist.

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

Empfohlene Korrektur

Entfernen Sie alle Verwendungen dieser Funktion.

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

Zusätzliche Empfehlungen¶

• Wenn Sie weitere Unterstützung benötigen, können Sie uns eine E-Mail an sma-support@snowflake.com senden oder ein Problem im SMA melden.`

Szenarien¶

Die Spark-Signatur für diese Methode DataFrameReader.load(path, format, schema, **options) existiert nicht in Snowpark. Daher wird jede Verwendung der Ladefunktion zu einer EWI im Ausgabecode führen.

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

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

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

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

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

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