迭代时证据参数的隐含值

Question

我的任务是从 jdbc 中获取表，然后将它们放到 s3 中。我已经使用 Slick 模式代码生成器生成了这些表的类。 如果我为每个表手动编写代码，它就可以完美运行。 就这样

Slick
  .source(Tables.table1.result)
  .runWith(ParquetStreams.toParquetSingleFile(s"s3a://bucket/table1"))
  .onComplete {
    case _ =>
      println("table1")
  }
Slick
  .source(Tables.table2.result)
  .runWith(ParquetStreams.toParquetSingleFile(s"s3a://bucket/table2"))
  .onComplete {
    case _ =>
      println("table2")
  }

问题是我有很多表，如果我可以迭代它们会更容易。

代码

val tbls = Map("table1" -> Tables.table1, "table2" -> Tables.table2)
tbls.foreach(table => {
  val table_name = table._1
  Slick
    .source(table_2.result)
    .runWith(ParquetStreams.toParquetSingleFile(s"s3a://bucket/$table_name"))
    .onComplete {
      case _ =>
        println(table_name)
    }
})

收到此错误。

could not find implicit value for evidence parameter of type com.github.mjakubowski84.parquet4s.ParquetRecordEncoder[_1#TableElementType]
[error]       .runWith(ParquetStreams.toParquetSingleFile(s"s3a://bucket/$table_name"))

编辑：------------------------------------------ ------------------

感谢您澄清事情。但即使有建议也行不通。应用这些东西后，存在同样的错误，并且出现了更多错误。

found   : slick.lifted.TableQuery[_1] where type _1 >: Tables.TableTwo with Tables.TableOne <: Tables.profile.Table[_ >: Tables.TableTwoRow with Tables.TableOneRow <: Product with java.io.Serializable]

按照某人的要求在此处留下简化代码。

import java.sql.Timestamp

import akka.actor.typed.ActorSystem
import akka.actor.typed.scaladsl.Behaviors
import akka.stream.alpakka.slick.javadsl.SlickSession
import akka.stream.alpakka.slick.scaladsl.Slick

import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContext}
import com.github.mjakubowski84.parquet4s.{ParquetRecordEncoder, ParquetSchemaResolver, ParquetStreams}

object Tables extends {
  val profile = slick.jdbc.MySQLProfile
} with Tables

/** Slick data model trait for extension, choice of backend or usage in the cake pattern. (Make sure to initialize this late.) */
trait Tables {
  val profile: slick.jdbc.JdbcProfile
  import profile.api._

  case class TableOneRow(id: Int, values: Option[String] = None)
  class TableOne(_tableTag: Tag) extends profile.api.Table[TableOneRow](_tableTag, Some("schema"), "table_one") {
    def * = (id, values) <> (TableOneRow.tupled, TableOneRow.unapply)

    val id: Rep[Int] = column[Int]("id", O.AutoInc, O.PrimaryKey)
    val values: Rep[Option[String]] = column[Option[String]]("values", O.Default(None))
  }
  lazy val TableOne = new TableQuery(tag => new TableOne(tag))

  case class TableTwoRow(id: Int, values: Option[Timestamp] = None)
  class TableTwo(_tableTag: Tag) extends profile.api.Table[TableTwoRow](_tableTag, Some("schema"), "table_two") {
    def * = (id, date) <> (TableTwoRow.tupled, TableTwoRow.unapply)

    val id: Rep[Int] = column[Int]("id", O.AutoInc, O.PrimaryKey)
    val date: Rep[Option[Timestamp]] = column[Option[Timestamp]]("date", O.Default(None))
  }
  lazy val TableTwo = new TableQuery(tag => new TableTwo(tag))
}

object Main extends App {

  implicit val actorSystem: ActorSystem[Nothing] = ActorSystem(Behaviors.empty, "alpakka-sample")
  implicit val executionContext: ExecutionContext = actorSystem.executionContext
  implicit val session = SlickSession.forConfig("slick-mysql") // (1)

  import session.profile.api._

  case class TableWithRecordEncoder[A](
    table: TableQuery[A])(
    implicit val recordEncoder: ParquetRecordEncoder[A]
  )

  def doTheThings[A](table: TableWithRecordEncoder[A], path: String) = {
    import table.recordEncoder
    Slick
      .source(table.table.result)
      .runWith(ParquetStreams.toParquetSingleFile(path))
      .onComplete {
        case _ =>
          println("Done. " + path)
      }
  }

  import polymorphic._
  def withRecordEncoder[A](table: TableQuery[A])(implicit recordEncoder: ParquetRecordEncoder[A])
  : Exists[TableWithRecordEncoder]
  = Exists(TableWithRecordEncoder(table))

  import polymorphic.syntax.all._
  Map("s3a://bucket/table_1" -> withRecordEncoder(Tables.TableOne),
     "s3a://bucket/table_2" -> withRecordEncoder(Tables.TableTwo)).
     foreach{ case (path, table) =>
      doTheThings(table.value, path)
    }

  actorSystem.whenTerminated.map(_ => session.close())
  Await.result(actorSystem.whenTerminated, Duration.Inf)

}

Answer 1

当您将Tables.table1 和Tables.table2 放入一个集合时，编译器会找到它们类型的最小上界，我们称之为T。 T 将是 Map 的值类型。但是隐式搜索是在编译时进行的，对于T#TableElementType没有合适的ParquetRecordEncoder实例，只针对具体的子类型。

因此，如果您想对这些进行迭代，您不仅需要将表添加到地图中，还需要添加所有相关的类型类实例。

//edit：这实际上有点棘手，所以我准备了一个例子来说明它是如何工作的。我希望你能从这里解决问题。

为了让事情变得更简单，我使用了polymorphic 库。

所以你有看起来像这样的Table 和ParquetRecordEncoder 类型：

trait Table[A] // details omitted
trait ParquetRecordEncoder[A]

我们需要一些实例来玩：

val intTable = new Table[Int] {}
val stringTable = new Table[String] {}
object ParquetRecordEncoder {
  implicit val int = new ParquetRecordEncoder[Int] {}
  implicit val string = new ParquetRecordEncoder[String] {}
}

现在我们需要一个类型来将这两者打包在一起：

case class TableWithRecordEncoder[A](
  table: Table[A])(
  implicit val recordEncoder: ParquetRecordEncoder[A]
)

我们需要一个函数来对这些对象做一些有用的事情：

  def doTheThings[A](table: TableWithRecordEncoder[A]) = {
    import table.recordEncoder // now we have a ParquetRecordEncoder[A]
                               // available as an implicit!

    ???                        // logic goes here
  }

但是由于每个表的类型参数A 会有所不同，我们不能将不同的TableWithRecordEncoder 对象填充到同一个Map 中！这就是来自polymorphic 库的Exists[F[_]] 类型的用武之地。它将所有这些不同的类型压缩为一种。让我们定义一个小助手来制作这样的对象：

import polymorphic._
def withRecordEncoder[A](table: Table[A])(implicit recordEncoder: ParquetRecordEncoder[A])
  : Exists[TableWithRecordEncoder]
  = Exists(TableWithRecordEncoder(table))

请注意，表格的元素类型根本不会出现在withRecordEncoder 的返回类型中。无论表格的元素类型是什么，该函数将始终返回一个Exists[TableWithRecordEncoder]！这很关键，因为现在我们可以在同一张地图中填充不同的表格：

import polymorphic.syntax.all._
  Map(
    "int" -> withRecordEncoder(intTable),
    "string" -> withRecordEncoder(stringTable)
  ).foreach { case (name, table) =>
    doTheThings(table.value)
  }

希望有帮助！