gpt4 book ai didi

scala - 如何动态定义流式数据集的模式以写入 csv?

转载 作者:行者123 更新时间:2023-12-02 01:10:27 26 4
gpt4 key购买 nike

我有一个流数据集,从 kafka 读取并尝试写入 CSV

case class Event(map: Map[String,String])
def decodeEvent(arrByte: Array[Byte]): Event = ...//some implementation
val eventDataset: Dataset[Event] = spark
.readStream
.format("kafka")
.load()
.select("value")
.as[Array[Byte]]
.map(decodeEvent)

Event 在内部保存 Map[String,String] 并写入 CSV 我需要一些模式。

假设所有字段都是 String 类型,所以我尝试了 spark repo 中的示例

val columns = List("year","month","date","topic","field1","field2")
val schema = new StructType() //Prepare schema programmatically
columns.foreach { field => schema.add(field, "string") }
val rowRdd = eventDataset.rdd.map { event => Row.fromSeq(
columns.map(c => event.getOrElse(c, "")
)}
val df = spark.sqlContext.createDataFrame(rowRdd, schema)

这会在运行时在“eventDataset.rdd”行上给出错误:

Caused by: org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;

下面不起作用,因为 '.map' 有一个列表 [String] 而不是元组

eventDataset.map(event => columns.map(c => event.getOrElse(c,""))
.toDF(columns:_*)

有没有办法通过编程模式和结构化流式数据集来实现这一点?

最佳答案

我会使用更简单的方法:

import org.apache.spark.sql.functions._

eventDataset.select(columns.map(
c => coalesce($"map".getItem(c), lit("")).alias(c)
): _*).writeStream.format("csv").start(path)

但如果你想要更接近当前解决方案的东西,请跳过 RDD 转换

import org.apache.spark.sql.catalyst.encoders.RowEncoder

eventDataset.rdd.map(event =>
Row.fromSeq(columns.map(c => event.getOrElse(c,"")))
)(RowEncoder(schema)).writeStream.format("csv").start(path)

关于scala - 如何动态定义流式数据集的模式以写入 csv?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45380479/

26 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com