Amazon EMR и S3, org.apache.spark.sql.AnalysisException: путь s3: /// var / table уже существует

Я пытаюсь найти источник ошибки в Spark 2.0.0, у меня есть карта, которая содержит имена таблиц в качестве ключей и фрейм данных в качестве значения, я просматриваю ее и в конце использую spark-avro (3.0.0 -preview2) для записи всего в каталоги S3. Он отлично работает локально (конечно, с локальным путем вместо пути s3), но когда я запускаю его на Amazon EMR, он работает некоторое время, а затем сообщает, что папка уже существует, и закрывается (что означало бы, что такое же значение ключа используется в этом цикле for более одного раза, верно?). Возможно, это проблема с потоковой передачей?

for ((k, v) <- tableMap) {
  val currTable: DataFrame = tableMap(k)
  val decryptedCurrTable = currTable.withColumn("data", decryptUDF(currTable("data")))
  val decryptedCurrTableData = sparkSession.sqlContext.read.json(decryptedCurrTable.select("data").rdd.map(row => row.toString()))
  decryptedCurrTable.write.avro(s"s3://..../$k/table")
  decryptedCurrTableData.write.avro(s"s3://..../$k/tableData")

amazon-s3 apache-spark amazon-emr spark-avro spark-dataframe
person Brady Auen    schedule 03.08.2016    source источник
comment
У вас есть журналы ошибок?   -  person error2007s    schedule 03.08.2016
comment
Да, но думаю, что ответил на свой вопрос. Я добавил .mode (добавить) после записи, и все сработало, так что это, должно быть, проблема параллелизма, imo.   -  person Brady Auen    schedule 03.08.2016

Ответы (1)


arrow_upward
3
arrow_downward

Я думаю, что это проблема параллелизма, я изменил свой код на:

decryptedCurrTable.write.mode("append").avro(s"s3://..../$k/table")
decryptedCurrTableData.write.mode("append").avro(s"s3://..../$k/tableData")

И все работало нормально.

person Brady Auen    schedule 03.08.2016
comment
Привет, я работал над аналогичным вариантом использования, но исходя из того, что я вижу в вашем собственном ответе, ваш ключ $ k продолжает меняться, что означает, что вы пишете не в одно и то же место назначения, поэтому режим добавления или переопределения не должен влиять на параллелизм ? - person c74ckds; 30.12.2019