scala如何参数化case类，并将case类变量传递给[T <：Product：TypeTag]

根据我对你的要求的理解，我认为以下应该是一个不错的起点。

def readDataset[A: Encoder](
spark: SparkSession,
mongoUrl: String,
collectionName: String,
clazz: Class[A]
): Dataset[A] = {
val config = ReadConfig(

Map("uri" -> s"$mongoUrl.$collectionName")

)

val df = MongoSpark.load(spark, config)

val fieldNames = clazz.getDeclaredFields.map(f => f.getName).dropRight(1).toList

val dfWithMatchingFieldNames = df.toDf(fieldNames: _*)

dfWithMatchingFieldNames.as[A]
}
你可以像这样使用它，

case class RsGoods(add_time: Int)

val spark: SparkSession = ...

import spark.implicts._

val rdGoodsDS = readDataset[RsGoods](
spark,
"mongodb://example.com/database",
"rs_goods",
classOf[RsGoods]
)
另外，以下两行，

val fieldNames = clazz.getDeclaredFields.map(f => f.getName).dropRight(1).toList

val dfWithMatchingFieldNames = df.toDf(fieldNames: _*)
只需要因为Spark通常用列名来读取DataFrames value1, value2, ...。所以我们想要更改列名以匹配我们的列名case class。

我不确定这些“defalut”列名是什么，因为MongoSpark参与其中。

您应该首先检查创建的df中的列名，如下所示，

val config = ReadConfig(
Map("uri" -> s"$mongoUrl.$collectionName")
)

val df = MongoSpark.load(spark, config)
如果，MongoSpark解决了这些“默认”列名称的问题并从您的集合中选择了库存名称，则不需要这两行，您的方法将变为这样，

def readDataset[A: Encoder](
spark: SparkSession,
mongoUrl: String,
collectionName: String,
): Dataset[A] = {
val config = ReadConfig(

Map("uri" -> s"$mongoUrl.$collectionName")

)

val df = MongoSpark.load(spark, config)

df.as[A]
}
和，

val rsGoodsDS = readDataset[RsGoods](
spark,
"mongodb://example.com/database",
"rs_goods"
)