使用Spark DataFrame在列上获取不同的值

Question

使用Spark 1.6.1版本我需要在列上获取不同的值,然后在其上执行一些特定的转换.该列包含超过5000万条记录,并且可以变大.
我知道做一个distinct.collect()会把呼叫带回驱动程序.目前我正在执行如下任务,是否有更好的方法？

 import sqlContext.implicits._
 preProcessedData.persist(StorageLevel.MEMORY_AND_DISK_2)

 preProcessedData.select(ApplicationId).distinct.collect().foreach(x => {
   val applicationId = x.getAs[String](ApplicationId)
   val selectedApplicationData = preProcessedData.filter($"$ApplicationId" === applicationId)
   // DO SOME TASK PER applicationId
 })

 preProcessedData.unpersist()  

Answer 1

那么要获得所有不同的值,Dataframe你可以使用distinct.正如您在文档中看到的那样,该方法返回另一个方法DataFrame.之后,您可以创建一个UDF以转换每个记录.

例如:

val df = sc.parallelize(Array((1, 2), (3, 4), (1, 6))).toDF("age", "salary")

// I obtain all different values. If you show you must see only {1, 3}
val distinctValuesDF = df.select(df("age")).distinct

// Define your udf. In this case I defined a simple function, but they can get complicated.
val myTransformationUDF = udf(value => value / 10)

// Run that transformation "over" your DataFrame
val afterTransformationDF = distinctValuesDF.select(myTransformationUDF(col("age")))

Answer 2

在 Pyspark 中尝试这个，

df.select('col_name').distinct().show()

Answer 3

该解决方案演示了如何使用比 UDF 更好的 Spark 本机函数转换数据。它还演示了哪个比某些查询dropDuplicates更适合。distinct

假设你有这个数据框：

+-------+-------------+
|country|    continent|
+-------+-------------+
|  china|         asia|
| brazil|south america|
| france|       europe|
|  china|         asia|
+-------+-------------+

以下是如何针对所有不同的国家/地区进行转型：

+-------+-------------+
|country|    continent|
+-------+-------------+
|  china|         asia|
| brazil|south america|
| france|       europe|
|  china|         asia|
+-------+-------------+

+--------------+
|       country|
+--------------+
|brazil is fun!|
|france is fun!|
| china is fun!|
+--------------+

dropDuplicates如果distinct您不想丢失信息，可以使用continent：

df
  .select("country")
  .distinct
  .withColumn("country", concat(col("country"), lit(" is fun!")))
  .show()

+-------+-------------+------------------------------------+
|country|continent    |description                         |
+-------+-------------+------------------------------------+
|brazil |south america|brazil is a country in south america|
|france |europe       |france is a country in europe       |
|china  |asia         |china is a country in asia          |
+-------+-------------+------------------------------------+

有关过滤 DataFrame 的更多信息，请参阅此处，有关删除重复项的更多信息，请参阅此处。

最终，您需要将转换逻辑包装在可以与 Dataset#transform 方法链接的自定义转换中。