Spark Structured Streaming MemoryStream + Row + Encoders 问题

Question

我正在尝试使用 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 中使用它更有魅力，没问题。

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

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

但是编译器抱怨如下：

没有找到参数 evidence$1 的隐含：Encoder[Row]

因此，我的问题是：我应该在这里做什么才能让它发挥作用

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

导入 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 中，我们还可以通过在 Seq 类型上运行 toDF 来利用控制台中 Spark 的隐式（如果您在 JAR 代码中导入它们）。这不适用于 null 类型，因此不一定推荐用于生产用例。

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

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

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

编辑1

我终于让它工作了

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

虽然我的问题是我对自己在做什么没有信心。在我看来，在某些情况下，我需要创建一个 DataSet 才能使用 toDF 转换将其转换为 DF[ROW]。我知道使用 DS 是类型安全的，但比使用 DF 慢。那么为什么这个中介有 DataSet 呢？这不是我第一次在 Spark Structured Streaming 中看到它。同样，如果有人可以帮助我解决这些问题，那就太好了。

Answer 1

我鼓励您使用 Scalacase classes进行数据建模。

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

现在，你可以有一个List的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")
  )

通过使用广为人知的抽象，结构化流 API可以很容易地推理流处理Dataset[T]。粗略地说，你只需要担心三件事：

• 源：源可以生成输入数据流，我们可以将其表示为Dataset[Input]. Input到达的每个新数据项都将被附加到这个无界数据集中。您可以随意操作数据（例如Dataset[Input]=> Dataset[Output]）。
• StreamingQueries和Sink：查询生成一个结果表，该表在每个触发间隔从 Source 更新。更改被写入称为 Sink 的外部存储中。
• 输出模式：您可以在不同的模式下将数据写入 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 时，您只需要定义 asource和 asink作为额外步骤。在这个例子中，我们可以使用 theMemoryStream和consolesink 来打印结果。

// 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()

}