找不到 AccessLog 类型的编码器。需要隐式 Encoder[AccessLog] 将 AccessLog 实例存储在数据集中

Posted 2023-04-15

【中文标题】找不到 AccessLog 类型的编码器。需要隐式 Encoder[AccessLog] 将 AccessLog 实例存储在数据集中

【英文标题】：Unable to find encoder for type AccessLog. An implicit Encoder[AccessLog] is needed to store AccessLog instances in a Dataset

【发布时间】：2021-05-29 12:41:26

【问题描述】：

您好，我正在解决 scala/spark 项目的问题，尝试进行一些计算，我的 scala 代码在 spark-shell 上运行良好，但是当尝试使用 sbt-assembly 运行相同的代码以将 scala 转换为 .jar 文件时，我面对这个错误：

。通过导入 spark.implicits 支持原始类型（Int、String 等）和产品类型（案例类）。未来版本中将添加对序列化其他类型的支持。

我正在尝试将 Dataset[List[String]] 转换为 Dataset[AccessLog] AccessLog 是一个案例类，通过映射使用。

Error screenshot

产生错误的代码：

import org.apache.spark.sql. Dataset, Encoder, SparkSession 
import org.apache.spark.sql.functions._

object DstiJob 

  // try and catch
  def run(spark: SparkSession, inputPath: String, outputPath: String): String = 
    // import spark.sqlContext.implicits._
    import spark.implicits._
    import org.apache.spark.sql. Encoder, Encoders 
    // implicit val enc: Encoder[AccessLog] = Encoders.product[AccessLog]

    val inputPath = "access.log.gz"
    val outputPath = "data/reports"
    val logsAsString = spark.read.text(inputPath).as[String]

    case class AccessLog(ip: String, ident: String, user: String, datetime: String, request: String, status: String, size: String, referer: String, userAgent: String, unk: String)

    val R = """^(?<ip>[0-9.]+) (?<identd>[^ ]) (?<user>[^ ]) \[(?<datetime>[^\]]+)\] \"(?<request>[^\"]*)\" (?<status>[^ ]*) (?<size>[^ ]*) \"(?<referer>[^\"]*)\" \"(?<useragent>[^\"]*)\" \"(?<unk>[^\"]*)\""".r
    val dsParsed = logsAsString.flatMap(x => R.unapplySeq(x))
    def toAccessLog(params: List[String]) = AccessLog(params(0), params(1), params(2), params(3), params(5), params(5), params(6), params(7), params(8), params(9))

    val ds: Dataset[AccessLog] = dsParsed.map(toAccessLog _)
    val dsWithTime = ds.withColumn("datetime", to_timestamp(ds("datetime"), "dd/MMM/yyyy:HH:mm:ss X"))
    dsWithTime.cache
    dsWithTime.createOrReplaceTempView("AccessLog")

【问题讨论】：

您是否阅读了错误消息，响应在消息中；）

@itIsNaz 它在 run 函数中。

【参考方案1】：

为了解决编译错误，case类应该定义在方法run之外。

而不是

object DstiJob 

    def run(spark: SparkSession, ...) 
       [...]
       case class AccessLog(...)
       val ds: Dataset[AccessLog] = ...
       [...]
    

你可以使用

object DstiJob 

   case class AccessLog(...)

   def run(spark: SparkSession, ...) 
       [...]  
       val ds: Dataset[AccessLog] = ...
       [...]
   

这应该可以解决问题，但不幸的是我无法解释为什么这会有所帮助。