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

Question

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

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

Answer 1

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

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

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")

火花是你的火花会议.