如何在另一个数据帧的 UDF 中引用数据帧？
How to reference a dataframe when in an UDF on another dataframe?
在另一个数据帧上执行 UDF 时如何引用 pyspark 数据帧？
这是一个虚拟示例。我正在创建两个数据帧 scores 和 lastnames，并且在每个数据帧中都有一个在两个数据帧中相同的列。在应用于 scores 的 UDF 中，我想过滤 lastnames 和 return 在 lastname.
中找到的字符串
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext
from pyspark.sql.types import *
sc = SparkContext("local")
sqlCtx = SQLContext(sc)
# Generate Random Data
import itertools
import random
student_ids = ['student1', 'student2', 'student3']
subjects = ['Math', 'Biology', 'Chemistry', 'Physics']
random.seed(1)
data = []
for (student_id, subject) in itertools.product(student_ids, subjects):
data.append((student_id, subject, random.randint(0, 100)))
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
schema = StructType([
StructField("student_id", StringType(), nullable=False),
StructField("subject", StringType(), nullable=False),
StructField("score", IntegerType(), nullable=False)
])
# Create DataFrame
rdd = sc.parallelize(data)
scores = sqlCtx.createDataFrame(rdd, schema)
# create another dataframe
last_name = ["Granger", "Weasley", "Potter"]
data2 = []
for i in range(len(student_ids)):
data2.append((student_ids[i], last_name[i]))
schema = StructType([
StructField("student_id", StringType(), nullable=False),
StructField("last_name", StringType(), nullable=False)
])
rdd = sc.parallelize(data2)
lastnames = sqlCtx.createDataFrame(rdd, schema)
scores.show()
lastnames.show()
from pyspark.sql.functions import udf
def getLastName(sid):
tmp_df = lastnames.filter(lastnames.student_id == sid)
return tmp_df.last_name
getLastName_udf = udf(getLastName, StringType())
scores.withColumn("last_name", getLastName_udf("student_id")).show(10)
下面是跟踪的最后一部分：
Py4JError: An error occurred while calling o114.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:335)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:344)
at py4j.Gateway.invoke(Gateway.java:252)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:209)
at java.lang.Thread.run(Thread.java:745)
将配对更改为字典以便于查找名称
data2 = {}
for i in range(len(student_ids)):
data2[student_ids[i]] = last_name[i]
而不是创建 rdd 并使其成为 df 创建广播变量
//rdd = sc.parallelize(data2)
//lastnames = sqlCtx.createDataFrame(rdd, schema)
lastnames = sc.broadcast(data2)
现在在广播变量 (lastnames) 上使用 values attr 在 udf 中访问它。
from pyspark.sql.functions import udf
def getLastName(sid):
return lastnames.value[sid]
您不能直接从 UDF 内部引用数据帧（或 RDD）。 DataFrame 对象是您的驱动程序的句柄，spark 使用它来表示将在集群上发生的数据和操作。 UDF 中的代码将 运行 在 Spark 选择的时间出现在集群上。 Spark 通过序列化该代码、复制闭包中包含的任何变量并将它们发送给每个工作人员来实现这一点。
相反，您想要做的是使用 Spark 在 API 到 join/combine 两个 DataFrame 中提供的结构。如果其中一个数据集很小，您可以手动发送广播变量中的数据，然后从您的 UDF 访问它。否则，您可以像以前一样创建两个数据框，然后使用连接操作将它们组合起来。这样的事情应该有效：
joined = scores.withColumnRenamed("student_id", "join_id")
joined = joined.join(lastnames, joined.join_id == lastnames.student_id)\
.drop("join_id")
joined.show()
+---------+-----+----------+---------+
| subject|score|student_id|last_name|
+---------+-----+----------+---------+
| Math| 13| student1| Granger|
| Biology| 85| student1| Granger|
|Chemistry| 77| student1| Granger|
| Physics| 25| student1| Granger|
| Math| 50| student2| Weasley|
| Biology| 45| student2| Weasley|
|Chemistry| 65| student2| Weasley|
| Physics| 79| student2| Weasley|
| Math| 9| student3| Potter|
| Biology| 2| student3| Potter|
|Chemistry| 84| student3| Potter|
| Physics| 43| student3| Potter|
+---------+-----+----------+---------+
同样值得注意的是，Spark DataFrames 在底层进行了优化，如果它足够小，可以将作为连接一部分的 DataFrame 转换为广播变量以避免随机播放。因此，如果您执行上面列出的连接方法，您应该获得最佳性能，而不会牺牲处理更大数据集的能力。
在另一个数据帧上执行 UDF 时如何引用 pyspark 数据帧？
这是一个虚拟示例。我正在创建两个数据帧 scores 和 lastnames，并且在每个数据帧中都有一个在两个数据帧中相同的列。在应用于 scores 的 UDF 中，我想过滤 lastnames 和 return 在 lastname.
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext
from pyspark.sql.types import *
sc = SparkContext("local")
sqlCtx = SQLContext(sc)
# Generate Random Data
import itertools
import random
student_ids = ['student1', 'student2', 'student3']
subjects = ['Math', 'Biology', 'Chemistry', 'Physics']
random.seed(1)
data = []
for (student_id, subject) in itertools.product(student_ids, subjects):
data.append((student_id, subject, random.randint(0, 100)))
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
schema = StructType([
StructField("student_id", StringType(), nullable=False),
StructField("subject", StringType(), nullable=False),
StructField("score", IntegerType(), nullable=False)
])
# Create DataFrame
rdd = sc.parallelize(data)
scores = sqlCtx.createDataFrame(rdd, schema)
# create another dataframe
last_name = ["Granger", "Weasley", "Potter"]
data2 = []
for i in range(len(student_ids)):
data2.append((student_ids[i], last_name[i]))
schema = StructType([
StructField("student_id", StringType(), nullable=False),
StructField("last_name", StringType(), nullable=False)
])
rdd = sc.parallelize(data2)
lastnames = sqlCtx.createDataFrame(rdd, schema)
scores.show()
lastnames.show()
from pyspark.sql.functions import udf
def getLastName(sid):
tmp_df = lastnames.filter(lastnames.student_id == sid)
return tmp_df.last_name
getLastName_udf = udf(getLastName, StringType())
scores.withColumn("last_name", getLastName_udf("student_id")).show(10)
下面是跟踪的最后一部分：
Py4JError: An error occurred while calling o114.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:335)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:344)
at py4j.Gateway.invoke(Gateway.java:252)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:209)
at java.lang.Thread.run(Thread.java:745)
将配对更改为字典以便于查找名称
data2 = {}
for i in range(len(student_ids)):
data2[student_ids[i]] = last_name[i]
而不是创建 rdd 并使其成为 df 创建广播变量
//rdd = sc.parallelize(data2)
//lastnames = sqlCtx.createDataFrame(rdd, schema)
lastnames = sc.broadcast(data2)
现在在广播变量 (lastnames) 上使用 values attr 在 udf 中访问它。
from pyspark.sql.functions import udf
def getLastName(sid):
return lastnames.value[sid]
您不能直接从 UDF 内部引用数据帧（或 RDD）。 DataFrame 对象是您的驱动程序的句柄，spark 使用它来表示将在集群上发生的数据和操作。 UDF 中的代码将 运行 在 Spark 选择的时间出现在集群上。 Spark 通过序列化该代码、复制闭包中包含的任何变量并将它们发送给每个工作人员来实现这一点。
相反，您想要做的是使用 Spark 在 API 到 join/combine 两个 DataFrame 中提供的结构。如果其中一个数据集很小，您可以手动发送广播变量中的数据，然后从您的 UDF 访问它。否则，您可以像以前一样创建两个数据框，然后使用连接操作将它们组合起来。这样的事情应该有效：
joined = scores.withColumnRenamed("student_id", "join_id")
joined = joined.join(lastnames, joined.join_id == lastnames.student_id)\
.drop("join_id")
joined.show()
+---------+-----+----------+---------+
| subject|score|student_id|last_name|
+---------+-----+----------+---------+
| Math| 13| student1| Granger|
| Biology| 85| student1| Granger|
|Chemistry| 77| student1| Granger|
| Physics| 25| student1| Granger|
| Math| 50| student2| Weasley|
| Biology| 45| student2| Weasley|
|Chemistry| 65| student2| Weasley|
| Physics| 79| student2| Weasley|
| Math| 9| student3| Potter|
| Biology| 2| student3| Potter|
|Chemistry| 84| student3| Potter|
| Physics| 43| student3| Potter|
+---------+-----+----------+---------+
同样值得注意的是，Spark DataFrames 在底层进行了优化，如果它足够小，可以将作为连接一部分的 DataFrame 转换为广播变量以避免随机播放。因此，如果您执行上面列出的连接方法，您应该获得最佳性能，而不会牺牲处理更大数据集的能力。