我使用 Spark 1.3 中的 DataFrame API。
我想从 DataFrame 中的日期获取星期几，而不会丢失 DataFrame 的所有元素。
在使用 DataFrame API 之前，我曾经使用 jodatime 在简单的地图上获取它。
目前可行的解决方案：
sqlContext.createDataFrame(myDataFrame.map(l=>operationOnTheField(l)),myDataFrame.schema))
是否可以在不返回RDD[Row] 上的地图的情况下执行该操作，然后使用此 RDD 创建一个 DataFrame？
【问题讨论】：
标签： scala apache-spark apache-spark-sql
您可以结合使用在 DataFrame 上调用 select() 和用户定义函数 (UDF) 来转换相关列。
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.functions._
用于设置示例DataFrame 的案例类。
private case class Cust(id: Integer, name: String,
sales: Double, discount: Double, state: String)
然后设置SQLContext并创建DataFrame如下：
import sqlContext.implicits._
val custs = Seq(
Cust(1, "Widget Co", 120000.00, 0.00, "AZ"),
Cust(2, "Acme Widgets", 410500.00, 500.00, "CA"),
Cust(3, "Widgetry", 410500.00, 200.00, "CA"),
Cust(4, "Widgets R Us", 410500.00, 0.0, "CA"),
Cust(5, "Ye Olde Widgete", 500.00, 0.0, "MA")
)
val customerDF = sc.parallelize(custs, 4).toDF()
注册一个简单的 UDF，您将使用它来转换“折扣”列。
val myFunc = udf {(x: Double) => x + 1}
获取列，已将 UDF 应用于“折扣”列，其他列保持不变。
val colNames = customerDF.columns
val cols = colNames.map(cName => customerDF.col(cName))
val theColumn = customerDF("discount")
我想找到一种“更好”的方式来匹配该列，但以下方法有效。
使用as() 为该列命名只是因为我们可以！
val mappedCols = cols.map(c =>
if (c.toString() == theColumn.toString()) myFunc(c).as("transformed") else c)
使用 select() 生成新的DataFrame
val newDF = customerDF.select(mappedCols:_*)
你变了
id name sales discount state
1 Widget Co 120000.0 0.0 AZ
2 Acme Widgets 410500.0 500.0 CA
3 Widgetry 410500.0 200.0 CA
4 Widgets R Us 410500.0 0.0 CA
5 Ye Olde Widgete 500.0 0.0 MA
进入
id name sales transformed state
1 Widget Co 120000.0 1.0 AZ
2 Acme Widgets 410500.0 501.0 CA
3 Widgetry 410500.0 201.0 CA
4 Widgets R Us 410500.0 1.0 CA
5 Ye Olde Widgete 500.0 1.0 MA
您可以找到完整的示例source code here。如果您对精确的列替换不挑剔，则可以使其更简单。
【讨论】：
试试这个
Table.select(Table("Otherkey"),MyUdf(Table("ColNeeded")).as("UdfTransformed"))
MyUdf 是你定义的 udf。
【讨论】：