scala - 在我停止工作之前,Spark Structured Streaming writestream 不会写入文件
问题描述
我在一个经典用例上使用 Spark Structured Streaming:我想从 kafka 主题中读取数据并将流以 parquet 格式写入 HDFS。
这是我的代码:
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{ArrayType, DataTypes, StructType}
object TestKafkaReader extends App{
val spark = SparkSession
.builder
.appName("Spark-Kafka-Integration")
.master("local")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
import spark.implicits._
val kafkaDf = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers","KAFKA_BROKER_IP:PORT")
//.option("subscribe", "test")
.option("subscribe", "test")
.option("startingOffsets", "earliest")
.load()
val moviesJsonDf = kafkaDf.selectExpr("CAST(value AS STRING)")
// movie struct
val struct = new StructType()
.add("title", DataTypes.StringType)
.add("year", DataTypes.IntegerType)
.add("cast", ArrayType(DataTypes.StringType))
.add("genres", ArrayType(DataTypes.StringType))
val moviesNestedDf = moviesJsonDf.select(from_json($"value", struct).as("movie"))
// json flatten
val movieFlattenedDf = moviesNestedDf.selectExpr("movie.title", "movie.year", "movie.cast","movie.genres")
// convert to parquet and save to hdfs
val query = movieFlattenedDf
.writeStream
.outputMode("append")
.format("parquet")
.queryName("movies")
.option("checkpointLocation", "src/main/resources/chkpoint_dir")
.start("src/main/resources/output")
.awaitTermination()
}
语境 :
- 我直接从 intellij 运行它(安装了本地 spark)
- 我设法毫无问题地从kafka读取并在控制台中写入(使用控制台模式)
- 目前我想在本地机器上写文件(但我确实在 HDFS 集群上尝试过,问题是一样的)
我的问题 :
在工作期间,它不会在文件夹中写入任何内容,我必须手动停止工作才能最终看到文件。
我认为这可能与.awaitTermination()
信息有关,我试图删除此选项,但如果没有该选项,我会收到错误并且作业根本无法运行。
也许我没有设置正确的选项,但是在阅读了很多次文档并在 Google 上搜索后,我什么也没找到。
你能帮我吗?
谢谢
编辑 :
- 我正在使用火花 2.4.0
- 我尝试了 64/128mb 格式 => 在我停止工作之前没有任何改变没有文件
解决方案
是的问题解决
我的问题是,我的数据太少,spark 正在等待更多数据来写入镶木地板文件。
为了完成这项工作,我使用了@AlexandrosBiratsis 的评论(更改块大小)
再次感谢@AlexandrosBiratsis 非常感谢你
推荐阅读
- c# - 添加 Microsoft.ACE.OLEDB.12.0 作为参考/包含在部署中?
- javascript - 对象和嵌套对象生成
- swift - 用于在二维数组中查找岛的 DFS 函数返回不正确的数字
- product - magento 1.9 stock qty shipping message php code
- scala - 在Scala中将元组转换为数组
- node.js - 即使配置了环境,也找不到节点模块
- postgresql - 尝试将包含 timestampz 数组的列拆分为 Postgres 中的分隔文本字符串时出现“函数不存在”错误
- python - 你能在 python 中复制 matlab 的旧伪随机数生成器吗?
- networking - Istio 可以在延迟更高、丢包的环境中工作吗?
- r - 使用 ABC_mcmc 函数时的错误: