如何在数据范围中引用广播变量

Question

我使用spark1.6。我尝试广播RDD，不确定如何访问数据帧中的广播变量？

我有两个DataFrames员工＆amp;部门。

员工dataframe

-------------------
Emp Id | Emp Name | Emp_Age
------------------
1 | john | 25
2 | David | 35

部门DataFrame

--------------------
Dept Id | Dept Name | Emp Id
-----------------------------
1 | Admin | 1
2 | HR | 2
import scala.collection.Map
val df_emp = hiveContext.sql("select * from emp")
val df_dept = hiveContext.sql("select * from dept")
val rdd = df_emp.rdd.map(row => (row.getInt(0),row.getString(1)))
val lkp = rdd.collectAsMap()
val bc = sc.broadcast(lkp)
print(bc.value.get(1).get)
--Below statement doesn't work
val combinedDF = df_dept.withColumn("emp_name",bc.value.get($"emp_id").get)

1. 如何在上述组合语句中引用广播变量？
2. 如果LKP不返回任何值，如何处理？
3. 是否有一种方法可以从LKP返回多个记录（假设是否有2个emp_id = 1的记录，我想获得这两个记录）
4. 如何从广播中返回多个值...（emp_name＆amp; emp_age）

Answer 1

如何在上述组合语句中引用广播变量？

使用udf。如果emp_id是Int

val f = udf((emp_id: Int) =>  bc.value.get(emp_id))
df_dept.withColumn("emp_name", f($"emp_id"))

如果LKP不返回任何值，如何处理？

不要使用上述get

是否有一种方法可以从LKP

返回多个记录

使用groupByKey：

val lkp = rdd.groupByKey.collectAsMap()

和explode：

df_dept.withColumn("emp_name", f($"emp_id")).withColumn("emp_name", explode($"emp_name"))

或只是跳过所有步骤，broadcast ：

import org.apache.spark.sql.functions._
df_emp.join(broadcast(df_dep), Seq("Emp Id"), "left")