概念：ユーザー定義関数

ユーザー定義関数を使うと、PySpark で独自の任意の Python コードを使用できます。たとえば、UDF を使って、データセットの各行にある複雑なテキスト形式から情報をパースできます。

宣言後、UDF は concat、date_diff、trim などの PySpark の組み込み関数と同じように動作します。

背景

直感に反しますが、通常の状況では、データが実際に Python コードに取り込まれることはありません。PySpark で DataFrame を操作するときは、最終的な DataFrame を作成するために Spark クラスターが分散して並列に実行すべき手順を記述しています。この仕組みにより、Spark と Foundry はほぼ無限にスケールできますが、クラスター内で実際のデータに対して実行するコードを組み込むには、UDF の簡単な設定が必要になります。PySpark は、クエリを実行する各サーバーに UDF のコードを送信します。

UDF の代替手段を検討する

PySpark の組み込み関数で同じロジックを表現できる場合は、UDF の使用を避けてください。UDF は大幅に遅いためです。UDF を使用する必要がある場合は、ベストプラクティスについて PySpark スタイルガイドを参照してください。

例

Copied!
1
"Weather report: rain 55-62"

次の天気情報の形式から最低気温、この場合は 55 を取得したいとします。次のような通常の Python 関数を記述できます。

Copied!
1
2
def extract_low_temperature(weather_report):
    return int(weather_report.split(' ')[-1].split('-')[0])

extract_low_temperature 関数を PySpark クエリに組み込むには、この関数をラップする UDF を作成します。UDF を作成する際は、関数と、PySpark の型システムで期待される戻り値の型を指定します。

Copied!
1
2
3
4
5
# 必要なデータ型をインポートします
from pyspark.sql.types import IntegerType

# 関数を UDF としてラップします
low_temp_udf = F.udf(extract_low_temperature, IntegerType())

これで、列全体を引数として渡し、DataFrame で UDF を使用できます。

Copied!
1
df = df.withColumn('low', low_temp_udf(F.col('weather_report')))

複数の列から読み取る

UDF は任意の列を引数として受け取れます。列の引数は、関数の引数に対応します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
from pyspark.sql.types import StringType

def weather_quality(temperature, windy):
    if temperature > 70 and windy == False:
        return "good"
    else:
        return "bad"

weather_udf = F.udf(weather_quality, StringType())

df = df.withColumn('quality', weather_udf(F.col('temp'), F.col('wind')))

UDF をデバッグする

Spark は、UDF 内で書き込まれたログをトップレベルのドライバーログに含めません。UDF 内の値を確認するには、デバッグ情報を列として返す一時的な UDF を作成します。ほかの手法については、ログ記録を参照してください。