スパークデータフレームのルックアップ

Question

私はスパーク1.6を使用しています。データフレーム内でルックアップを実装する方法を知りたいと思います。

私は従業員&という2つのデータフレームを持っています。

• 従業員DATAFRAME

  ------------------- 
  Emp Id | Emp Name 
  ------------------ 
  1 | john 
  2 | David 
  

• 部門DATAFRAME

  -------------------- 
  Dept Id | Dept Name | Emp Id 
  ----------------------------- 
  1 | Admin | 1 
  2 | HR | 2 
  

私は、部門テーブルに従業員表からEMP IDを検索し、DEPT名を取得したいと思います。したがって、結果セットは

Emp Id | Dept Name 
------------------- 
1 | Admin 
2 | HR 

です。SPARKでこのルックアップUDF機能を実装するにはどうすればよいですか。私は両方のデータフレームでJOINを使用したくありません。

Answer 1

コメントで既に述べたように、データフレームに参加することは、方法です。

ルックアップを使用できますが、ルックアップテーブルをドライバメモリに収集する必要がある、「分散型」ソリューションはないと思います。

import org.apache.spark.sql.functions._ 
import sqlContext.implicits._ 
import scala.collection.Map 

val emp = Seq((1,"John"),(2,"David")) 
val deps = Seq((1,"Admin",1),(2,"HR",2)) 

val empRdd = sc.parallelize(emp) 
val depsDF = sc.parallelize(deps).toDF("DepID","Name","EmpID") 


val lookupMap = empRdd.collectAsMap() 
def lookup(lookupMap:Map[Int,String]) = udf((empID:Int) => lookupMap.get(empID)) 

val combinedDF = depsDF 
    .withColumn("empNames",lookup(lookupMap)($"EmpID")) 

私の最初に考えたのはUDFにempRddを渡し、PairRDDに定義されたlookupメソッドを使用していたが、あなたは火花を持つことができないので、これは当然のことながら動作しません。また、このアプローチは、EmpIDをが一意であることを前提としています変換（すなわち、UDF）内のアクション（すなわち、lookup）。

EDIT：

あなたempDfは、複数の列（例えば名前、年齢）を持っている場合、あなたはこの

val empRdd = empDf.rdd.map{row => 
     (row.getInt(0),(row.getString(1),row.getInt(2)))} 


    val lookupMap = empRdd.collectAsMap() 
    def lookup(lookupMap:Map[Int,(String,Int)]) = 
     udf((empID:Int) => lookupMap.lift(empID)) 

    depsDF 
     .withColumn("lookup",lookup(lookupMap)($"EmpID")) 
     .withColumn("empName",$"lookup._1") 
     .withColumn("empAge",$"lookup._2") 
     .drop($"lookup") 
     .show() 

Answer 2

あなたは既に次の手順を実行しデータフレームは、その非常に簡単持っていると言っているとおり：

1）sqlcontext

val sqlContext = new org.apache.spark.sql.SQLContext(sc) 

2を作成します）、すべての3例：用の一時テーブルを作成します。

EmployeeDataframe.createOrReplaceTempView("EmpTable") 

3）MySQLクエリを使用したクエリ

val MatchingDetails = sqlContext.sql("SELECT DISTINCT E.EmpID, DeptName FROM EmpTable E inner join DeptTable G on " + 
    "E.EmpID=g.EmpID")