При записи в S3, почему я получаю FileNotFoundException - PullRequest
Новая
       12

При записи в S3, почему я получаю FileNotFoundException

0 голосов
/ 13 февраля 2020

Я использую Spark- SQL -2.3.1, Kafka, Java 8 в своем проекте и хотел бы использовать AWS -S3 в качестве хранилища дикарей.

Я пишу / Сохранение потребленных данных из Kafka topi c в корзину S3, как показано ниже: 

   ds.writeStream()
     .format("parquet")
     .option("path", parquetFileName)
     .option("mergeSchema", true)
     .outputMode("append")
     .partitionBy("company_id")
     .option("checkpointLocation", checkPtLocation)
     .trigger(Trigger.ProcessingTime("25 seconds"))
     .start();

Но во время записи я получаю FileNotFoundException

Caused by: java.io.FileNotFoundException: No such file or directory: s3a://company_id=216231245/part-00055-f4f87dc9-a620-41bd-9380-de4ba7e70efb.c000.snappy.parquet
  at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:1931)
  at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:1822)
  at org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:1763)

Я понимаю, почему я ' получаю FileNotFoundException при написании? я не читаю с S3 не так ли? Так что здесь происходит и как это исправить?

1 Ответ

2 голосов
/ 13 февраля 2020

Это потому, что S3 не файловая система, а хранилище объектов. Он не поддерживает семантику, необходимую для переименования, как HDFS. Сначала Spark записывает выходные файлы во временную папку, а затем переименовывает их. В S3 нет атомарного способа сделать это. Вот почему иногда вы увидите эти ошибки.

Теперь, чтобы исправить это, если ваша среда позволяет, вы можете использовать HDFS в качестве промежуточного хранилища и переместить файлы на S3 для последующей обработки.

Если у вас включено oop 3.1 Вы можете использовать коммиттеры s3a, поставляемые с ним. Более подробную информацию о том, как настроить это, можно найти здесь

Если вы используете более старую версию had oop, вы можете использовать выходной коммиттер S3 для Spark, который в основном использует multi -часть загрузки в mimi c это переименование. Один такой коммиттер, о котором я знаю, это это . Похоже, это не обновляется в последнее время, хотя. Могут быть и другие варианты.

...