You need to say .mode("append") if you want to append to existing data.On Tue, Dec 22, 2015 at 6:48 AM, Yash Sharma <yash...@gmail.com> wrote: > Well you are right. Having a quick glance at the source[1] I see that the > path creation does not consider the partitions. > > It tries to create the path before looking for partitions columns. > > Not sure what would be the best way to incorporate it. Probably you can > file a jira and experienced contributors can share their thoughts. > > 1. > https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ResolvedDataSource.scala > > Line- 131 > > - Thanks, via mobile, excuse brevity. > On Dec 22, 2015 7:48 PM, "Jan Holmberg" <jan.holmb...@perigeum.fi> wrote: > >> In my example directories were distinct. >> >> So If I would like to have to distinct directories ex. >> >> /tmp/data/year=2012 >> /tmp/data/year=2013 >> >> It does not work with >> val df = Seq((2012, "Batman")).toDF("year","title") >> >> df.write.partitionBy("year").avro("/tmp/data") >> >> val df2 = Seq((2013, "Batman")).toDF("year","title") >> >> df2.write.partitionBy("year").avro("/tmp/data") >> >> >> As you can see, it complains about the target directory (/tmp/data) and >> not about the partitioning keys. >> >> >> org.apache.spark.sql.AnalysisException: path hdfs://nameservice1/tmp/data >> already exists.; >> at >> org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation.run(InsertIntoHadoopFsRelation.scala:76) >> at >> org.apache.spark.sql.execution.ExecutedCommand.sideEffectResult$lzycompute(commands.scala:57) >> at >> org.apache.spark.sql.execution.ExecutedCommand.sideEffectResult(commands.scala:57) >> at >> org.apache.spark.sql.execution.ExecutedCommand.doExecute(commands.scala:69) >> at >> org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140) >> >> >> >> On 22 Dec 2015, at 15:44, Yash Sharma <yash...@gmail.com> wrote: >> >> Well this will indeed hit the error if the next run has similar year and >> months and writing would not be possible. >> >> You can try working around by introducing a runCount in partition or in >> the output path. >> >> Something like- >> >> /tmp/data/year/month/01 >> /tmp/data/year/month/02 >> >> Or, >> /tmp/data/01/year/month >> /tmp/data/02/year/month >> >> This is a work around. >> >> Am sure other better approaches would follow. >> >> - Thanks, via mobile, excuse brevity. >> On Dec 22, 2015 7:01 PM, "Jan Holmberg" <jan.holmb...@perigeum.fi> wrote: >> >>> Hi Yash, >>> >>> the error is caused by the fact that first run creates the base >>> directory ie. "/tmp/data" and the second batch stumbles to the existing >>> base directory. I understand that the existing base directory is a >>> challenge but I do not understand how to make this work with streaming >>> example where each batch would create a new distinct directory. >>> >>> Granularity has no impact. No matter how data is partitioned, second >>> 'batch' always fails with existing base dir. >>> >>> scala> df2.write.partitionBy("year").avro("/tmp/data") >>> org.apache.spark.sql.AnalysisException: path >>> hdfs://nameservice1/tmp/data already exists.; >>> at >>> org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation.run(InsertIntoHadoopFsRelation.scala:76) >>> at >>> org.apache.spark.sql.execution.ExecutedCommand.sideEffectResult$lzycompute(commands.scala:57) >>> at >>> org.apache.spark.sql.execution.ExecutedCommand.sideEffectResult(commands.scala:57) >>> at >>> org.apache.spark.sql.execution.ExecutedCommand.doExecute(commands.scala:69) >>> at >>> org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140) >>> at >>> org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138) >>> at >>> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147) >>> at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138) >>> at >>> org.apache.spark.sql.SQLContext$QueryExecution.toRdd$lzycompute(SQLContext.scala:933) >>> at >>> org.apache.spark.sql.SQLContext$QueryExecution.toRdd(SQLContext.scala:933) >>> at >>> org.apache.spark.sql.execution.datasources.ResolvedDataSource$.apply(ResolvedDataSource.scala:197) >>> at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:146) >>> at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:137) >>> at >>> com.databricks.spark.avro.package$AvroDataFrameWriter$$anonfun$avro$1.apply(package.scala:37) >>> at >>> com.databricks.spark.avro.package$AvroDataFrameWriter$$anonfun$avro$1.apply(package.scala:37) >>> at >>> $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:33) >>> at >>> $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:38) >>> >>> >>> On 22 Dec 2015, at 14:06, Yash Sharma <yash...@gmail.com> wrote: >>> >>> Hi Jan, >>> Is the error because a past run of the job has already written to the >>> location? >>> >>> In that case you can add more granularity with 'time' along with year >>> and month. That should give you a distinct path for every run. >>> >>> Let us know if it helps or if i missed anything. >>> >>> Goodluck >>> >>> - Thanks, via mobile, excuse brevity. >>> On Dec 22, 2015 2:31 PM, "Jan Holmberg" <jan.holmb...@perigeum.fi> >>> wrote: >>> >>>> Hi, >>>> I'm stuck with writing partitioned data to hdfs. Example below ends up >>>> with 'already exists' -error. >>>> >>>> I'm wondering how to handle streaming use case. >>>> >>>> What is the intended way to write streaming data to hdfs? What am I >>>> missing? >>>> >>>> cheers, >>>> -jan >>>> >>>> >>>> import com.databricks.spark.avro._ >>>> >>>> import org.apache.spark.sql.SQLContext >>>> >>>> val sqlContext = new SQLContext(sc) >>>> >>>> import sqlContext.implicits._ >>>> >>>> val df = Seq( >>>> (2012, 8, "Batman", 9.8), >>>> (2012, 8, "Hero", 8.7), >>>> (2012, 7, "Robot", 5.5), >>>> (2011, 7, "Git", 2.0)).toDF("year", "month", "title", "rating") >>>> >>>> df.write.partitionBy("year", "month").avro("/tmp/data") >>>> >>>> val df2 = Seq( >>>> (2012, 10, "Batman", 9.8), >>>> (2012, 10, "Hero", 8.7), >>>> (2012, 9, "Robot", 5.5), >>>> (2011, 9, "Git", 2.0)).toDF("year", "month", "title", "rating") >>>> >>>> df2.write.partitionBy("year", "month").avro("/tmp/data") >>>> --------------------------------------------------------------------- >>>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org >>>> For additional commands, e-mail: user-h...@spark.apache.org >>>> >>>> >>> >>
