Search code examples
scalaapache-sparkparquetdatabricksspark-structured-streaming

Cannot write a stream into a parquet sink

I work on Databricks, a data processing platform based on Spark with an HDFS-like file system, so basically I beleive those of you who are familiar with Spark and HDFS will be able to help me without Databricks experience.

I read data from Kafka using Structured Streaming:

var streamingInputDF = 
  spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<XX.XX.XXX.XX:9092")
    .option("subscribe", "answers")     
    .option("startingOffsets", "earliest")  
    .option("minPartitions", "1")  
    .option("failOnDataLoss", "true")
    .load()

Do some transformations:

 val streamingSelectDF = streamingInputDF
  .withWatermark("timestamp","1 days")
  .select(explode(split($"value".cast("string"), "\\s+")).as("word"), col("timestamp"))
  .groupBy(window($"timestamp", "1 minute"), $"word")
  .count
  .where("count >= 11")

Then I make sure there is some data by printing to console:

+--------------------+----+-----+
|              window|word|count|
+--------------------+----+-----+
|[2019-06-10 14:33...| the|  763|
|[2019-06-09 20:48...| the|  523|
|[2019-06-10 14:33...| and|  489|
|[2019-06-10 14:33...|   a|  479|
|[2019-06-08 19:07...| the|  435|
|[2019-06-10 14:33...|  to|  430|
|[2019-06-10 14:33...|  of|  365|
|[2019-06-09 20:48...|   a|  314|
|[2019-06-09 20:48...| and|  303|
|[2019-06-09 20:48...|  to|  285|
|[2019-06-10 14:33...|  is|  272|
|[2019-06-08 19:07...|   a|  264|
|[2019-06-08 19:07...| and|  250|
|[2019-06-08 19:07...|  to|  233|
|[2019-06-09 20:48...|  of|  231|
|[2019-06-10 14:33...|  in|  219|
|[2019-06-10 14:33...|that|  211|
|[2019-06-08 19:07...|  of|  186|
|[2019-06-10 14:33...| for|  166|
|[2019-06-09 20:48...|  is|  158|
+--------------------+----+-----+
only showing top 20 rows

Then, I want to stream the data into a parquet file:

val query =
  streamingSelectDF
    .writeStream
    .format("parquet")
    .option("path", "/mnt/kafka/answers")
    .option("checkpointLocation", "/mnt/kafka/checkpoint")
    .partitionBy("window")
    .start()

But no files are created. Only a "kafka" directory is created:

ls /mnt/

path               name     size
dbfs:/mnt/kafka/   kafka/   0
Solution

  • I've solved the problem.

    I used the /mnt/ path which is a location that is kept for connecting a blob storage.

    Since I wanted to store the parquets on DBFS, I had to use any path except for /mnt/ so I changed it to /tmp/ instead.