火花从一个小文件中查找答案

【问题标题】：spark look up from a small file火花从一个小文件中查找
【发布时间】：2017-11-11 11:53:23
【问题描述】：

我正在做一个 spark 项目，需要建议如何以最佳方式解决以下问题：

我有一个数据框（比如 MainDF ），它有数百万条记录。格式是这样的 (name:String, value:Int) 。内容示例如下：

Davi,130
Joel,20
Emma,500

我有另一个小文件，有 4 行记录，像这样 (className:String,minValue:Int,maxValue:Int) 现在我需要根据 min 和 max 之间的值查找类名来创建一个文件，输出以下记录：

First,500,9999999
Second,100,499
Third,0,99
Unknown,-99999,0

我需要为 MainDF 中的每个值查找这个小文件，并根据 small File.Example 中的值范围添加类名称：

Davi,130,Second
Joel,20,Third
Emma,500,First

这是我写的代码：

//Main Data read, millions of records
val MainData = sc.textFile("/mainfile.csv")
case class MainType(Name:String,value:Int)
val MainDF = MainData .map(line => line.split(",")).map(e =>MainType(e(0),e(1).toInt))).toDF
MainDF.registerTempTable("MainTable")
val refData = sc.broadast( sc.textFile("/refdata.csv"))
case class refDataType (className:String,minValue:Int,maxValue:Int)
//ref data, just 4 records
val refRDD = refData.map(line=> line.split(",")).map( e => refDataType ( e(0) , e(1).toInt, e(2).toInt))

我想我必须在这里写一个 UDF，但我不知道如何在 UDF 中使用 Dataframe，或者在 spark scala 中有什么方法可以做到这一点

【问题讨论】：

如果这 4 个条件总是相同，您可以编写 if -else if -else UDF，而不是从该文件创建数据框
@philantrovert 感谢您的关注，用户可能会对其进行更新，因此可能会有所不同。
这里除了使用UDF还有其他方法吗？
@philantrovert，谢谢。让我试试这个。

标签： scala apache-spark user-defined-functions lookup

【解决方案1】：

您可以使用textFile 将文件作为RDD 读取，收集它，因为它非常小（可能根据您的要求进行广播）。

通过收集 RDD 获得数组后，您可以创建 Range 和 UDF 以检查您的值是否在该范围内。

val rdd = sc.parallelize(Array(
("First",500,9999999),
("Second",100,499),
("Third",0,99),
("Unknown",-99999,0)
))

val dataArr = rdd.map{ case (className, min, max) => 
                       (className, Range(min, max) )  }.collect
// First Element will be the Class Name
// Second will be the Range(min, max)
// sc.broadcast(dataArr) here

val getClassName = udf {(x: Int) => { 
                  dataArr.map{ e => 
                        if (e._2.contains(x) ) e._1.toString 
                        else null.asInstanceOf[String] }
                  .filter(_ != null )
                  .apply(0) }}

df.withColumn("ClassName", getClassName($"VALUE") ).show
+----+-----+---------+
|NAME|VALUE|ClassName|
+----+-----+---------+
|Davi|  130|   Second|
|Joel|   20|    Third|
|Emma|  500|    First|
+----+-----+---------+

我很肯定可能会有更好的解决方案。

【讨论】：

【解决方案2】：

这里最简单的方法是使用 csv 数据源读取这两个文件并使用标准 SparkSQL 连接它们，如下所示：

import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}
val mainSchema = StructType(Seq(StructField("name", StringType, false), 
StructField("value", IntegerType, false)))
val mainDf = spark.read.schema(mainSchema).csv("/tmp/b.txt")
val lookupSchema = StructType(Seq(StructField("class_name", StringType, false), StructField("min_value", IntegerType, false), 
StructField("max_value", IntegerType, false)))
val lookupDf = spark.read.schema(lookupSchema).csv("/tmp/a.txt")
val result = mainDf.join(lookupDf, $"value" <= $"max_value" && $"value" > $"min_value")
result.show()

我不确定最高效的方式是这种方式还是@philantrovert 建议的方式（这也可能取决于您使用的 Spark 版本）。您应该尝试这两种方法并自己决定。

【讨论】：