将 dbfs 文件作为 databricks 中的流数据帧获取

Question

我有一个问题，我需要为每个CSV文件中的数据链球插入一个外部表，该文件将降落到ADLS Gen 2存储中。

当我从dbutils.fs.ls（）输出中获取流式数据框时，我考虑了一个解决方案，然后调用一个在foreachBatch（）内创建表的函数。

我已经准备好了功能，但是我无法找到一种将目录信息流到流数据框中的方法。有人知道如何实现这一目标吗？

Answer 1

请检查以下代码块。

package com.sparkbyexamples.spark.streaming
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object SparkStreamingFromDirectory {

  def main(args: Array[String]): Unit = {

    val spark:SparkSession = SparkSession.builder()
      .master("local[3]")
      .appName("SparkByExamples")
      .getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")

    val schema = StructType(
      List(
        StructField("Zipcode", IntegerType, true),
        
      )
    )

    val df = spark.readStream
      .schema(schema)
      .json("Your directory")

    df.printSchema()

    val groupDF = df.select("Zipcode")
        .groupBy("Zipcode").count()
    groupDF.printSchema()

    groupDF.writeStream
      .format("console")
      .outputMode("complete")
      .start()
      .awaitTermination()
  }
}