PySparkのexpr関数について

概要

expr関数の引数内で利用できる関数はorg.apache.spark.sql.functionsに定義されているものだけでなく、下記のSpark SQL built-in functionsも利用可能である。
https://spark.apache.org/docs/latest/api/sql/

使用例

例1

下記ではsqrtを用いている。
https://spark.apache.org/docs/latest/api/sql/#sqrt

xy_df = xy_df.withColumn('distance',F.expr('sqrt(x * x + y * y)'))

例2

randなどを用いて、-1〜1の乱数をセット
https://spark.apache.org/docs/latest/api/sql/#rand

rand_df = rand_df.withColumn('random',F.expr('rand() * 2 - 1'))

例3

上記のものなどを組み合わせて、モンテカルロ法の円周率
※Spark in Actionのmap/reduceとは別のやり方で。(パーフォマンスンは....)

darts = 10000

l = []

for x in range(darts):
    l.append([x])
#   l.append((x,)) tupleの場合
# https://jamiekt.wordpress.com/2016/12/13/creating-a-spark-dataframe-containing-only-one-column/

xy_df = spark.createDataFrame(l, ["index"])

xy_df = (
    xy_df
    .withColumn("x", F.expr("rand() * 2 - 1"))
    .withColumn("y", F.expr("rand() * 2 - 1"))
    .withColumn("distance", F.expr("sqrt(x * x + y * y)"))
    .withColumn("inside_a_circle", F.expr("CASE WHEN distance < 1 THEN True ELSE False END")
    )
)

4 * xy_df.filter('inside_a_circle = True').count() / 10000