Spark 结构化流内存流 + 行 + 编码器问题

我正在尝试使用 Spark 结构化流在本地计算机上运行一些测试。

在批处理模式下，这是我正在处理的行：

val recordSchema = StructType(List(StructField("Record", MapType(StringType, StringType), false)))
val rows         = List(
    Row(
      Map("ID" -> "1",
        "STRUCTUREID" -> "MFCD00869853",
        "MOLFILE" -> "The MOL Data",
        "MOLWEIGHT" -> "803.482",
        "FORMULA" -> "C44H69NO12",
        "NAME" -> "Tacrolimus",
        "HASH" -> "52b966c551cfe0fa7d526bac16abcb7be8b8867d",
        "SMILES" -> """[H][C@]12O[C@](O)([C@H](C)C[C@@H]1OC)""",
        "METABOLISM" -> "The metabolism 500"
       )),
    Row(
      Map("ID" -> "2",
        "STRUCTUREID" -> "MFCD00869854",
        "MOLFILE" -> "The MOL Data",
        "MOLWEIGHT" -> "603.482",
        "FORMULA" -> "",
        "NAME" -> "Tacrolimus2",
        "HASH" -> "52b966c551cfe0fa7d526bac16abcb7be8b8867d",
        "SMILES" -> """[H][C@]12O[C@](O)([C@H](C)C[C@@H]1OC)""",
        "METABOLISM" -> "The metabolism 500"
      ))
  )
val df  = spark.createDataFrame(spark.sparkContext.parallelize(rows), recordSchema)

在 Batch more 中使用它是一种魅力，没有问题。

现在我尝试使用 MemoryStream 进入流模式进行测试。我添加了以下内容：

implicit val ctx = spark.sqlContext
val intsInput = MemoryStream[Row]

但编译器抱怨如下：

未找到参数证据$1的隐式：编码器[行]

因此，我的问题是：我应该在这里做什么才能使其正常工作

我还看到，如果添加以下导入，错误就会消失：

导入spark.implicits._

实际上，我现在收到以下警告而不是错误

参数证据 $1 的不明确隐式：编码器 [行]

我不太了解编码器机制，如果有人可以向我解释如何不使用这些隐式，我将不胜感激。原因是当涉及到从 Rows 创建 DataFrame 时，我在一本书中红色了以下内容。

推荐方法：

val myManualSchema = new StructType(Array(
  new StructField("some", StringType, true),
  new StructField("col", StringType, true),
  new StructField("names", LongType, false)))
val myRows = Seq(Row("Hello", null, 1L))
val myRDD = spark.sparkContext.parallelize(myRows)
val myDf = spark.createDataFrame(myRDD, myManualSchema)
myDf.show()

然后作者继续这样说：

在Scala中，我们还可以利用Spark的隐含功能 控制台（如果您将它们导入 JAR 代码中），方法是在 序列类型。这不适用于空类型，所以它不是 必须推荐用于生产用例。

val myDF = Seq(("Hello", 2, 1L)).toDF("col1", "col2", "col3")

如果有人可以花时间解释当我使用隐式时在我的场景中发生了什么，并且这样做是否相当安全，或者是否有一种方法可以更明确地做到这一点而不导入隐式。

最后，如果有人能给我指点关于编码器和 Spark 类型映射的好文档，那就太好了。

EDIT1

我终于可以使用它了

  implicit val ctx = spark.sqlContext
  import spark.implicits._
  val rows = MemoryStream[Map[String,String]]
  val df = rows.toDF()

尽管我的问题是我对自己所做的事情没有信心。在我看来，就像在某些情况下我需要创建一个 DataSet 才能将其转换为 DF[ROW] 并进行 toDF 转换。我知道使用 DS 是类型安全的，但比使用 DF 慢。那么为什么要使用 DataSet 这个中介呢？这不是我第一次在 Spark 结构化流中看到这一点。再说一次，如果有人能帮助我解决这些问题，那就太好了。

我鼓励你使用 Scalacase classes用于数据建模。

final case class Product(name: String, catalogNumber: String, cas: String, formula: String, weight: Double, mld: String)

现在你可以拥有一个List of Product在记忆中：

  val inMemoryRecords: List[Product] = List(
    Product("Cyclohexanecarboxylic acid", " D19706", "1148027-03-5", "C(11)H(13)Cl(2)NO(5)", 310.131, "MFCD11226417"),
    Product("Tacrolimus", "G51159", "104987-11-3", "C(44)H(69)NO(12)", 804.018, "MFCD00869853"),
    Product("Methanol", "T57494", "173310-45-7", "C(8)H(8)Cl(2)O", 191.055, "MFCD27756662")
  )

The 结构化流API https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html通过使用众所周知的方法可以很容易地推理流处理Dataset[T]抽象。粗略地说，您只需要担心三件事：

• Source https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources：源可以生成输入数据流，我们可以将其表示为Dataset[Input]。每个新数据项Input到达的数据将被附加到这个无界数据集中。您可以根据需要操纵数据（例如Dataset[Input] => Dataset[Output]).
• 流式查询 https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#starting-streaming-queries and Sink https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-sinks：查询生成一个结果表，该结果表在每个触发间隔从源更新。更改被写入称为接收器的外部存储中。
• 输出方式 https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes：可以通过不同的模式将数据写入 Sink：完整模式、追加模式和更新模式。

假设您想了解分子量大于 200 单位的产品。

正如您所说，使用批处理 API 相当简单直接：

// Create an static dataset using the in-memory data
val staticData: Dataset[Product] = spark.createDataset(inMemoryRecords)

// Processing...
val result: Dataset[Product] = staticData.filter(_.weight > 200)

// Print results!
result.show()

使用 Streaming API 时，您只需要定义一个source and a sink作为额外的步骤。在这个例子中，我们可以使用MemoryStream和console水槽打印结果。

// Create an streaming dataset using the in-memory data (memory source)
val productSource = MemoryStream[Product]
productSource.addData(inMemoryRecords)

val streamingData: Dataset[Product] = productSource.toDS()

// Processing...
val result: Dataset[Product] = streamingData.filter(_.weight > 200)

// Print results by using the console sink. 
val query: StreamingQuery = result.writeStream.format("console").start()

// Stop streaming
query.awaitTermination(timeoutMs=5000)
query.stop()

请注意，staticData和streamingData具有确切的类型签名（即Dataset[Product]）。这使得我们无论使用 Batch 还是 Streaming API 都可以应用相同的处理步骤。您还可以考虑实现一个通用方法def processing[In, Out](inputData: Dataset[In]): Dataset[Out] = ???以避免在这两种方法中重复自己。

完整代码示例：

object ExMemoryStream extends App {

  // Boilerplate code...
  val spark: SparkSession = SparkSession.builder
    .appName("ExMemoryStreaming")
    .master("local[*]")
    .getOrCreate()

  spark.sparkContext.setLogLevel("ERROR")

  import spark.implicits._
  implicit val sqlContext: SQLContext = spark.sqlContext

  // Define your data models 
  final case class Product(name: String, catalogNumber: String, cas: String, formula: String, weight: Double, mld: String)

  // Create some in-memory instances
  val inMemoryRecords: List[Product] = List(
    Product("Cyclohexanecarboxylic acid", " D19706", "1148027-03-5", "C(11)H(13)Cl(2)NO(5)", 310.131, "MFCD11226417"),
    Product("Tacrolimus", "G51159", "104987-11-3", "C(44)H(69)NO(12)", 804.018, "MFCD00869853"),
    Product("Methanol", "T57494", "173310-45-7", "C(8)H(8)Cl(2)O", 191.055, "MFCD27756662")
  )

  // Defining processing step
  def processing(inputData: Dataset[Product]): Dataset[Product] =
    inputData.filter(_.weight > 200)

  // STATIC DATASET
  val datasetStatic: Dataset[Product] = spark.createDataset(inMemoryRecords)

  println("This is the static dataset:")
  processing(datasetStatic).show()

  // STREAMING DATASET
  val productSource = MemoryStream[Product]
  productSource.addData(inMemoryRecords)

  val datasetStreaming: Dataset[Product] = productSource.toDS()

  println("This is the streaming dataset:")
  val query: StreamingQuery = processing(datasetStreaming).writeStream.format("console").start()
  query.awaitTermination(timeoutMs=5000)
  
  // Stop query and close Spark
  query.stop()
  spark.close()

}