【问题标题】：PySpark and broadcast join examplePySpark 和广播连接示例
【发布时间】：2016-03-07 07:37:28
【问题描述】：

我使用的是 Spark 1.3

# Read from text file, parse it and then do some basic filtering to get   data1
data1.registerTempTable('data1')

# Read from text file, parse it and then do some basic filtering to get data1
data2.registerTempTable('data2')

# Perform join
data_joined = data1.join(data2, data1.id == data2.id);

我的数据非常倾斜，data2 (几 KB)

【问题讨论】：

标签： python apache-spark apache-spark-sql pyspark

【解决方案1】：

Spark 1.3 不支持使用 DataFrame 进行广播连接。在 Spark >= 1.5.0 中，您可以使用 broadcast 函数来应用广播连接：

from pyspark.sql.functions import broadcast

data1.join(broadcast(data2), data1.id == data2.id)

对于旧版本，唯一的选择是转换为 RDD 并应用与其他语言相同的逻辑。大致是这样的：

from pyspark.sql import Row
from pyspark.sql.types import StructType

# Create a dictionary where keys are join keys
# and values are lists of rows
data2_bd = sc.broadcast(
    data2.map(lambda r: (r.id, r)).groupByKey().collectAsMap())


# Define a new row with fields from both DFs
output_row = Row(*data1.columns + data2.columns)

# And an output schema
output_schema = StructType(data1.schema.fields + data2.schema.fields)

# Given row x, extract a list of corresponding rows from broadcast
# and output a list of merged rows
def gen_rows(x):
    return [output_row(*x + y) for y in data2_bd.value.get(x.id, [])]

# flatMap and create a new data frame
joined = data1.rdd.flatMap(lambda row: gen_rows(row)).toDF(output_schema)

【讨论】：

pyspark.sql.functions.broadcast最早出现在1.6，据the docs
@NicholasWhite In PySpark 包装器已在 1.6 中添加，但 Scala 方法从 1.5 开始可用，因此您也可以使其在 1.5 中工作。

【解决方案2】：

此代码适用于 spark-2.0.2-bin-hadoop2.7 版本

from pyspark.sql import SparkSession

from pyspark.sql.functions import broadcast

spark = SparkSession.builder.appName("Python Spark SQL basic 
example").config("spark.some.config.option", "some-value").getOrCreate()

df2 = spark.read.csv("D:\\trans_mar.txt",sep="^");

df1=spark.read.csv("D:\\trans_feb.txt",sep="^");

print(df1.join(broadcast(df2),df2._c77==df1._c77).take(10))

【讨论】：