Невозможно записать поток в паркетную раковину - PullRequest
Новая
       33

Невозможно записать поток в паркетную раковину

1 голос
/ 14 июня 2019

Я работаю над Databricks, платформой обработки данных на основе Spark с HDFS-подобной файловой системой, поэтому в основном я верю, что те из вас, кто знаком со Spark и HDFS, смогут помочь мне без опыта работы с Databricks.

Я читаю данные из Kafka, используя структурированную потоковую передачу: 

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

Выполните некоторые преобразования: 

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

Затем я проверяю, есть ли какие-то данные, печатая на консоль: 

+--------------------+----+-----+
|              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

Затем я хочу передать данные в файл паркета: 

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

Но файлы не создаются.Создается только каталог "kafka": 

ls /mnt/

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

1 Ответ

1 голос
/ 15 июня 2019

Я решил проблему.

Я использовал / mnt / path, который хранится для подключения хранилища больших двоичных объектов.

Поскольку я хотел хранить паркет в DBFS, мне пришлось использовать любой путь, кроме / mnt /, поэтому я изменил его на / tmp /.

...