将火花数据帧动态转换为元组数据集（字符串，_<：产品）

Posted 2023-04-15

【中文标题】将火花数据帧动态转换为元组数据集（字符串，_<：产品）

【英文标题】：Dyamically cast spark dataframe to dataset of tuple(String,_<:Product)

【发布时间】：2019-11-09 21:02:38

【问题描述】：

我在使用 spark 时遇到了一个特殊的问题，我不太确定发生了什么，如果有人可以提供帮助，那就太好了。我的问题是有一个类似于下面的功能，即将数据帧转换为某种类型的数据集，这是在运行时决定的。我需要使用数据集，因为底层案例类有一些我想使用的注释。

 def ret(spark: SparkSession, dss: DataFrame, typ: String): Dataset[_ <: Product] = 
    import spark.implicits._
    typ match 
      case "t1" => dss.as[T1]
      case "t2" => dss.as[T2]
    

  

我可以使用以下函数调用 val ds = ret(spark,dataframe,"t1") 将数据帧转换为数据集

这个函数一切正常，现在我想扩展现有函数以返回Dataset[(String,_<:Product)]，所以我像这样修改我的函数，

 def ret(spark: SparkSession, dss: DataFrame,typ: String):Dataset[(String,_ <: Product)] = 
    import spark.implicits._
    typ match 
      case "t1" => dss.as[(String,T1)]
      case "t2" => dss.as[(String,T2)]
    
  

这给了我一个编译错误说，类型(String,T1)，与预期类型(String,_<:Product) 不匹配。这里实际发生了什么？有什么想法可以解决这个问题吗？任何提示将不胜感激！

非常感谢！！

更新：上界<: product scala.product t1 case>

case class T1(name: String, age: Int)

case class T2(name: String, max: Int, min: Int)

但它可以是任何东西

【问题讨论】：

你能用 T1、T2 和产品的定义更新你的问题吗？

@ijayadeep 感谢您的评论，完成。

【参考方案1】：

Dataset[(String, T1)] 和Dataset[(String, T2)] 的共同超类型不是Dataset[(String,_ <: Product)]，而是更复杂的existential type

Dataset[(String, T)] forSome  type T <: Product 

Dataset[(String,_ <: Product)] 也确实是一种存在类型，但不同；它是

的简写

Dataset[(String, T) forSome  type T <: Product ]

请注意，要在没有警告的情况下使用Dataset[(String, T)] forSome type T <: Product ，您需要添加import scala.language.existentials（并且这些类型将是removed in Scala 3）。

编辑：我认为我检查的内容就足够了，但显然类型推断在这里失败了，我真的不明白为什么。

def ret(spark: SparkSession, dss: DataFrame, typ: String): Dataset[(String, T)] forSome  type T <: Product  = 
  import spark.implicits._
  typ match 
    case "t1" => dss.as[(String,T1)]: (Dataset[(String, T)] forSome  type T <: Product )
    case "t2" => dss.as[(String,T2)]: (Dataset[(String, T)] forSome  type T <: Product )
  

确实按预期编译。您可以提取类型别名以避免重复：

type DatasetStringT = Dataset[(String, T)] forSome  type T <: Product 

def ret(spark: SparkSession, dss: DataFrame, typ: String): DatasetStringT = 
  import spark.implicits._
  typ match 
    case "t1" => dss.as[(String,T1)]: DatasetStringT 
    case "t2" => dss.as[(String,T2)]: DatasetStringT 
  

【讨论】：

非常感谢您的回答，但我仍然得到编译 Error:(70, 114) type mismatch; found : org.apache.spark.sql.Dataset[_1] where type _1 >: (String, T1) with (String, T2) <: (String, Product with Serializable) required: org.apache.spark.sql.Dataset[(String, T)] forSome type T <: Product Note: _1 <: (String, Product), but class Dataset is invariant in type T. You may wish to define T as +T instead. (SLS 4.5) def ret(spark: SparkSession, dss: DataFrame, typ: String):Dataset[(String, T)] forSome type T <: Product = 知道我在这里可能做错了什么吗？

这是因为 Dataset API 是不变的。这意味着如果Something是不变的，那么即使A是B的子类，那么Something[A]也不是Something[B]的子类。如果您可以解释您尝试实现的用例，则可能有其他实现方式，因为您将无法修改 Dataset API

@jjayadeep 是的，但Dataset[(String, T)] forSome type T <: Product 不是Dataset[Something]。

语句表示模式匹配返回一个type DatasetStringT，上面已经定义了。它不是强制转换，而是一种确保代码符合类型系统的方法。

@SaiKiranKrishnaMurthy 正如 jjayadeep 所说。它也可以触发隐式转换，但这里没有。技术术语是“类型归属”，您可以在此处阅读更多内容***.com/questions/2087250/…