Snowpark Connect for Spark プロパティ¶

コメント¶

Snowflake共通テーブル式（CTEs） を有効にする設定。この設定により、多くの反復コードブロックが存在するSnowflakeクエリが最適化されます。この変更により、クエリのコンパイルと実行の両方のパフォーマンスが向上します。

コメント¶

true に設定した場合、特定の操作の動作を変更します。

• bit_get/getbit --- Snowflake getbit関数 の明示的な使用

• hash --- Snowflakeハッシュ関数 の明示的な使用

• md5 --- Snowflake md5関数 の明示的な使用

• テーブル列の名前変更 --- テーブル列を変更できるようにします

コメント¶

デフォルトでは、 Snowpark Connect for Spark は、すべての整数型を Long 型として扱います。これは、 Snowflakeで数値が表現される 方法に起因します。統合型のエミュレートにより、データソースからの読み取り時に、Snowpark型とSpark型間の正確なマッピングが可能になります。

デフォルトのオプション client_default はScalaクライアントからスクリプトが実行されたときにのみエミュレートをアクティブ化します。統合型は、次の精度に基づいてマップされます。

他の精度が見つかった場合、最終的な型は DecimalType にマッピングされます。

コメント¶

提供されたディレクトリからパーティション化されたファイルの正確なサブセットのみを読み取るには、追加の設定が必要です。この機能は、 外部ステージ に保存されたファイルでのみ使用できます。読み取りファイルを整理するために、 Snowpark Connect for Spark は 外部テーブル を使用します。

この機能は、構成 snowpark.connect.sql.partition.external_table_location が設定されている場合に有効化されます。外部テーブルが作成される既存のデータベース名とスキーマ名が含まれている必要があります。

外部ステージに保存されているParquetファイルを読み取ると、外部テーブルが作成されます。内部ステージ上のファイルの場合は作成されません。スキーマを提供すると、実行時間が短縮され、ソースからスキーマを推論するコストが削減されます。

最高のパフォーマンスを得るには、 Snowflake外部テーブルフィルタリングの制限 に従ってフィルターしてください。

spark.conf.set("snowpark.connect.sql.partition.external_table_location", "<database-name>.<schema-name>")

spark.read.parquet("@external-stage/example").filter(col("x") > lit(1)).show()

schema = StructType([StructField("x",IntegerType()),StructField("y",DoubleType())])

spark.read.schema(schema).parquet("@external-stage/example").filter(col("x") > lit(1)).show()

コメント¶

この構成は、 snowpark.connect.udf.python.imports と非常によく似ています。これを使用すると、 `registerJavaFunction<https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.UDFRegistration.registerJavaFunction.html>`_ を使用して作成されたJava UDFs の外部ライブラリやファイルを指定できます。構成は、不要な依存関係の混在を防ぐために、相互に排他的です。

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

# Files need to be previously staged

spark.conf.set("snowpark.connect.udf.java.imports", "[@stage/library.jar]")

spark.registerJavaFunction("javaFunction", "com.example.ExampleFunction")

spark.sql("SELECT javaFunction('arg')").show()

コメント¶

これを使って、Python UDFs で利用可能な追加パッケージを定義できます。値は依存関係のコンマ区切りリストです。

Snowflakeで次の SQL を実行すると、サポートされているシステムパッケージのリストを見つけることができます。

SELECT * FROM INFORMATION_SCHEMA.PACKAGES WHERE LANGUAGE = 'python';

spark.conf.set("snowpark.connect.udf.packages", "[numpy]")

@udtf(returnType="val: int")

class Powers:

  def eval(self, x: int):
      import numpy as np

      for v in np.power(np.array([x, x, x]), [0, 1, 2]):
          yield (int(v),)

spark.udtf.register(name="powers", f=Powers)

spark.sql("SELECT * FROM powers(10)").show()

コメント¶

このプロパティは、 UDTFs がSpark互換の動作を使用するか、またはデフォルトのSnowpark動作を使用するかを決定します。true に設定した場合、Sparkの出力型の強制とエラー処理パターンを模倣する互換性ラッパーを適用します。

有効にすると、 UDTFs はSparkスタイルの自動型強制（例: 文字列「true」をブール値に、ブール値を整数に）とエラー処理を適用する互換性ラッパーを使用します。ラッパーは、位置アクセスと名前アクセスの両方においてテーブル引数を行のようなオブジェクトに変換し、Sparkの動作パターンに合わせて SQL null値を適切に処理します。

コメント¶

Snowflake は、列名の重複をサポートしていません。

df = spark.createDataFrame([
(1, 1),
(2, 2)
], ["foo", "foo"])

df.show() # works

df.createTempView("df_view") # Fails with SQL compilation error: duplicate column name 'foo'

コメント¶

この構成は、 snowpark.connect.udf.python.imports と非常によく似ています。これを使用して、 `registerJavaFunction<https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.UDFRegistration.registerJavaFunction.html>`_ を使用して作成されたJava UDFs の外部ライブラリやファイルを指定できます。構成は、不要な依存関係の混在を防ぐために、相互に排他的です。

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

# Files need to be previously staged

spark.conf.set("snowpark.connect.udf.java.imports", "[@stage/library.jar]")

spark.registerJavaFunction("javaFunction", "com.example.ExampleFunction")

spark.sql("SELECT javaFunction('arg')").show()

コメント¶

設定により、Python UDFs で利用可能な追加パッケージを定義できます。値は依存関係のコンマ区切りリストです。

Snowflakeで次の SQL を実行すると、サポートされているシステムパッケージのリストを見つけることができます。

SELECT * FROM INFORMATION_SCHEMA.PACKAGES WHERE LANGUAGE = 'python';

spark.conf.set("snowpark.connect.udf.packages", "[numpy]")

@udtf(returnType="val: int")

class Powers:

  def eval(self, x: int):
      import numpy as np

      for v in np.power(np.array([x, x, x]), [0, 1, 2]):
          yield (int(v),)

spark.udtf.register(name="powers", f=Powers)

spark.sql("SELECT * FROM powers(10)").show()

コメント¶

この設定は、 UDTFs がSpark互換の動作を使用するか、またはデフォルトのSnowpark動作を使用するかを決定します。有効にすると（true）、Sparkの出力型の強制（例: 文字列「true」をブール値に、ブール値を整数に）とエラー処理パターンを模倣する互換性ラッパーが適用されます。

ラッパーは、位置アクセスと名前付きアクセスの両方においてテーブル引数を行のようなオブジェクトに変換し、Sparkの動作パターンに合わせて SQL null値を適切に処理します。

コメント¶

Snowflakeテーブルはユーザー定義のパーティション分割をサポートしておらず、デフォルトでは、パーティションの上書きはエラーになります。このオプションを有効にすると、INSERT OVERWRITE <table> PARTITION(<partition spec>) 構文を使用して上書きを実行できます。

<partition spec> はターゲットテーブルに存在する列を受け入れます。

spark.conf.set("snowpark.connect.sql.emulatePartitionOverwritesForSnowflakeTables", True)

# create the students and persons tables as standard Snowflake tables
students_data = [
  ("Ashua Hill", "456 Erica Ct, Cupertino", 111111),
  ("Brian Reed", "723 Kern Ave, Palo Alto", 222222)
]

students_df = spark.createDataFrame(students_data, ["name", "address", "student_id"])
students_df.write.mode("overwrite").saveAsTable("students")

persons_data = [
    ("Dora Williams", "134 Forest Ave, Menlo Park", 123456789),
    ("Eddie Davis", "245 Market St, Milpitas", 345678901)
]

persons_df = spark.createDataFrame(persons_data, ["name", "address", "ssn"])
persons_df.write.mode("overwrite").saveAsTable("persons")

# overwrites all rows in the students table that have a student_id of 222222
spark.sql("""
    INSERT OVERWRITE students PARTITION (student_id = 222222)
    SELECT name, address FROM persons WHERE name = 'Dora Williams'
""").collect()

コメント¶

デフォルトでは、|spconnect|は、SnowflakeのキュレーションされたAnacondaチャネルからPythonパッケージを解決します。この構成をアーティファクトリポジトリ名に設定することで、PyPIまたはその他の構成済みソースからパッケージを解決できるようになり、Anacondaチャネルでは提供されていないパッケージを使用することが可能になります。

Snowflakeでアーティファクトリポジトリを作成および構成する方法については、:doc:`/developer-guide/udf/python/udf-python-packages`をご参照ください。

この構成を変更すると、キャッシュされたUDFsおよびUDTFsは無効化され、次回の呼び出し時に新しいリポジトリで再作成されることになります。

この構成は、以下の操作に適用されます。

• :code:`@udf`デコレーターまたは:code:`spark.udf.register()`により登録されたUDFs

• :code:`@udtf`デコレーターまたは:code:`spark.udtf.register()`により登録されたUDTFs

• groupBy().applyInPandas()`による:code:`applyInPandas

• DataFrame.mapInArrow()`による:code:`mapInArrow

• groupBy().cogroup().applyInPandas()`による:code:`cogroup

spark.conf.set("snowpark.connect.artifact_repository", "my_pypi_repo")
spark.conf.set("snowpark.connect.udf.packages", "[pykalman]")

@udf(returnType=DoubleType())
def kalman_smooth_value(value: float) -> float:
    import numpy as np
    from pykalman import KalmanFilter

    kf = KalmanFilter(
        transition_matrices=[1],
        observation_matrices=[1],
        initial_state_mean=0,
        initial_state_covariance=1,
        observation_covariance=1,
        transition_covariance=0.1,
    )
    observations = np.array([value, value, value])
    smoothed_state_means, _ = kf.smooth(observations)
    return float(smoothed_state_means[-1][0])

df = spark.createDataFrame([(1, 10.0), (2, 20.0), (3, 30.0)], ["id", "value"])
df.select("id", kalman_smooth_value("value").alias("smoothed")).show()

コメント¶

一部のサードパーティPythonパッケージ（TensorFlow､XGBoost、および特定の科学ライブラリなど）は、x86 CPUアーキテクチャ専用にビルドされています。この構成を:code:`x86`に設定すると、|spconnect|が生成する:code:`CREATE FUNCTION`ステートメントに:code:`RESOURCE_CONSTRAINT=(architecture='x86')`が追加され、UDFがx86互換のインフラストラクチャ上で実行されるようになります。

この構成を使用するには、x86リソース制約で作成されたウェアハウスでワークロードを実行する必要があります。次のリソース制約値は、x86をサポートしています。

• :code:`MEMORY_1X_x86`（最小ウェアハウスサイズ：XSMALL）

• :code:`MEMORY_16X_x86`（最小ウェアハウスサイズ：MEDIUM）

• :code:`MEMORY_64X_x86`（最小ウェアハウスサイズ：LARGE）

ウェアハウスにx86リソース制約がない場合、UDFの実行は失敗します。

この構成は、以下の操作に適用されます。

• :code:`@udf`デコレーターまたは:code:`spark.udf.register()`により登録されたUDFs

• :code:`@udtf`デコレーターまたは:code:`spark.udtf.register()`により登録されたUDTFs

• groupBy().applyInPandas()`による:code:`applyInPandas

CREATE WAREHOUSE my_x86_warehouse WITH
  WAREHOUSE_SIZE = 'MEDIUM'
  WAREHOUSE_TYPE = 'SNOWPARK-OPTIMIZED'
  RESOURCE_CONSTRAINT = 'MEMORY_16X_x86';

USE WAREHOUSE my_x86_warehouse;

spark.conf.set("snowpark.connect.udf.resource_constraint.architecture", "x86")

@udf(returnType=IntegerType())
def add_one(x: int) -> int:
    return x + 1

df = spark.createDataFrame([(1,), (2,), (3,)], ["value"])
df.select(add_one(df["value"]).alias("result")).show()