如何使用Spark Structured Streaming连续监视目录

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了如何使用Spark Structured Streaming连续监视目录相关的知识,希望对你有一定的参考价值。

我希望spark能够连续监视目录并在文件出现在该目录中时使用spark.readStream读取CSV文件。

请不要包含Spark Streaming的解决方案。我正在寻找一种方法来使用火花结构流。

答案

以下是此用例的完整解决方案案例:

如果您以独立模式运行。您可以将驱动程序内存增加为:

bin/spark-shell --driver-memory 4G

无需像独立模式一样设置执行程序内存执行程序在驱动程序中运行。

在完成@ T.Gaweda的解决方案时,找到以下解决方案:

val userSchema = new StructType().add("name", "string").add("age", "integer")
val csvDF = spark
  .readStream
  .option("sep", ";")
  .schema(userSchema)      // Specify schema of the csv files
  .csv("/path/to/directory")    // Equivalent to format("csv").load("/path/to/directory")

csvDf.writeStream.format("console").option("truncate","false").start()

现在,spark会持续监视指定的目录,只要在目录中添加任何csv文件,就会在该文件上执行DataFrame操作“csvDF”。

注意:如果您想要火花推测,您必须先设置以下配置:

spark.sqlContext.setConf("spark.sql.streaming.schemaInferenc‌​e","true")

火花是你的火花会议。

另一答案

正如官方documentation所写,你应该使用“文件”来源:

文件来源 - 将目录中写入的文件作为数据流读取。支持的文件格式为text,csv,json,parquet。有关更新的列表,请参阅DataStreamReader接口的文档,以及每种文件格式支持的选项。请注意,文件必须原子地放置在给定目录中,在大多数文件系统中,可以通过文件移动操作来实现。

从文档中获取的代码示例:

// Read all the csv files written atomically in a directory
val userSchema = new StructType().add("name", "string").add("age", "integer")
val csvDF = spark
  .readStream
  .option("sep", ";")
  .schema(userSchema)      // Specify schema of the csv files
  .csv("/path/to/directory")    // Equivalent to format("csv").load("/path/to/directory")

如果您未指定触发器,Spark将尽快读取新文件

以上是关于如何使用Spark Structured Streaming连续监视目录的主要内容,如果未能解决你的问题,请参考以下文章

如何使用 Python 在 Spark Structured Streaming 中查看特定指标

如何使用 Spark Structured Streaming 打印 Json 编码的消息

如何使用 Scala Case Class 在 Spark Structured Streaming 中映射 Kafka 源

如何在不使用 flatMapsGroupWithState 的情况下使用 Structured Streaming 2.3.0 在 spark 中进行无状态聚合?

如何在 Spark Structured Streaming 中控制输出文件的大小

Spark Structured Streaming - 如何按最新和聚合计数进行重复数据删除