ポイント: Apache SparkのためにPythonでユーザー定義関数（UDF）を書く方法を学び、データ処理を強化します。このチュートリアルでは、withColumn()、SQL文、RDDマップ変換の3つの方法を説明します。
- UDFにより、カスタム関数が登録され、Sparkデータフレームに適用されます。
- コード例がそれぞれの方法の実装を効果的に示しています。
この記事では、Apache SparkのためのPythonでユーザー定義関数（UDF）を書く方法について説明します。この例のコードはこちらです。
UDFが必要な理由
SparkはデータをデータフレームまたはRDD（弾力的分散データセット）に保存します。これをデータベースのように考えてください。従来のSQLデータベース（例えば、mySQL）と同様に、カスタム関数を作成してそれをデータベースに直接実行することはできません。最初に関数を登録する必要があります。つまり、それをデータベースに保存する必要があります。これは組み込みデータベース関数の一つであるかのように、sum()、average、count()などのようにです。
これがSparkデータフレームにおける状況です。Spark RDDでは、RDDの行に直接関数を実行できます。
UDFの3つのアプローチ
UDFを作成する方法は3つあります：
- df = df.withColumn
- df = sqlContext.sql("からのSQL文")
- rdd.map(customFunction())
以下に3つのアプローチを示します。最初から始めましょう。
アプローチ1：withColumn()
以下に、シンプルなデータフレームとRDDを作成します。データ構造内の唯一のテキストフィールドを整数に変換する関数を作成します。これは、たとえば、すべてのデータを数値に変換する必要がある機械学習を行っている場合に行うかもしれません。
以下のインポートに注意してください。各例でそれらを参照して、3つのアプローチのそれぞれに必要なオブジェクトをインポートする方法を知ってください。
以下はアプローチ1の完全なコードです。まず、重要なセクションを見てみましょう。通常の方法を使用してデータフレームを作成します：
df = spark.createDataFrame(data,schema=schema)
さて、私たちは2つのことを行います。まず、関数colsIntを作成し、それを登録します。その登録された関数はtoInt()という別の関数を呼び出しますが、これは登録する必要はありません。udf.register("colsInt", colsInt)の最初の引数は、関数を参照するために使用する名前です。2番目は、登録したい関数です。
colsInt = udf(lambda z: toInt(z), IntegerType())
spark.udf.register("colsInt", colsInt)
def toInt(s):
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
return(int(''.join(st)))
else:
return Null
次に、関数colinsIntを次のように呼び出します。最初の引数は、作成したい新しい列の名前です。2番目は、関数にプラグインするデータフレーム内の列です。
df2 = df.withColumn( 'semployee',colsInt('employee'))
注意してください。df['employees']は列オブジェクトであり、単一の従業員ではありません。つまり、その列のすべての行に対してループする必要があります。したがって、このラムダ（インライン）ループを使用します。
colsInt = udf(lambda z: toInt(z), IntegerType())
こちらがアプローチ1全体です：
import pyspark
from pyspark import SQLContext
from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType
from pyspark.sql.functions import udf
from pyspark.sql import Row
conf = pyspark.SparkConf()
sc = pyspark.SparkContext.getOrCreate(conf=conf)
spark = SQLContext(sc)
schema = StructType([
StructField("sales", FloatType(),True),
StructField("employee", StringType(),True),
StructField("ID", IntegerType(),True)
])
data = [[ 10.2, "Fred",123]]
df = spark.createDataFrame(data,schema=schema)
colsInt = udf(lambda z: toInt(z), IntegerType())
spark.udf.register("colsInt", colsInt)
def toInt(s):
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
return(int(''.join(st)))
else:
return Null
df2 = df.withColumn( 'semployee',colsInt('employee'))
今、結果を示します。新しい列semployeeが追加されたことに注意してください。withColumn()は新しいデータフレームを作成するので、df2を作成しました。
df2.show()
+-----+--------+---+----------+
|sales|employee| ID| semployee|
+-----+--------+---+----------+
| 10.2| Fred|123|1394624364|
+-----+--------+---+----------+
アプローチ2：SQLの使用
ここでの最初のステップは、データフレームをテーブルとして登録することです。これにより、SQL文を実行できるようになります。dfはデータフレームであり、dftabは作成する一時テーブルです。
spark.registerDataFrameAsTable(df, "dftab")
次に、既存のdfから新しいデータフレームdf3を作成し、employee列にcolsInt関数を適用します。
df3 = spark.sql("select sales, employee, ID, colsInt(employee) as iemployee from dftab")
結果は以下の通りです：
df3.show()
+-----+--------+---+----------+
|sales|employee| ID| iemployee|
+-----+--------+---+----------+
| 10.2| Fred|123|1394624364|
+-----+--------+---+----------+
アプローチ3：RDDマップ
データフレームにはmap()関数がありません。その関数を使用したい場合は、dff.rddを使用してデータフレームをRDDに変換する必要があります。
関数を次のように適用します：
rdd = df.rdd.map(toIntEmployee)
これは、行オブジェクトをtoIntEmployee関数に渡します。したがって、行オブジェクトを返す必要があります。RDDは不変であるため、新しい行を作成する必要があります。
以下では、そのフィールド内の各文字を整数に変換し、それらを連結します。
def toIntEmployee(rdd):
s = rdd["employee"]
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
e = int(''.join(st))
else:
e = s
return Row(rdd["sales"],rdd["employee"],rdd["ID"],e)
結果を印刷します：
for x in rdd.collect():
print(x)
Apache Sparkガイド
- Apache Sparkの機械学習パイプライン：入門
- Apache SparkによるSGD線形回帰の例
- Spark決定木分類器
- ScalaとSparkを使用したロジスティック回帰
- Apache SparkでのJupyterノートブックの使用方法
- PythonとSparkの機械学習を使用して分類を行う
- PythonでのSpark UDF（ユーザー定義関数）の書き方
- Python Spark ML K-Meansの例
- サイバーセキュリティに機械学習を適用する方法
- Apache Spark：ストリームの扱い
- Apache SparkによるK-meansクラスタリング
- Hiveと一緒にSparkを使用する
- 予防保守のための予測を行うためにApache Sparkを使用する方法
これらの投稿は私自身の意見であり、BMCの立場、戦略、または意見を必ずしも代表するものではありません。