spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jan Holmberg <jan.holmb...@perigeum.fi>
Subject Re: Writing partitioned Avro data to HDFS
Date Tue, 22 Dec 2015 21:33:33 GMT
Works like a charm. Thanks a lot!

-jan

On 22 Dec 2015, at 23:08, Michael Armbrust <michael@databricks.com<mailto:michael@databricks.com>>
wrote:

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 <yash360@gmail.com<mailto:yash360@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.holmberg@perigeum.fi<mailto:jan.holmberg@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 <yash360@gmail.com<mailto:yash360@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.holmberg@perigeum.fi<mailto:jan.holmberg@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 <yash360@gmail.com<mailto:yash360@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.holmberg@perigeum.fi<mailto:jan.holmberg@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-unsubscribe@spark.apache.org<mailto:user-unsubscribe@spark.apache.org>
For additional commands, e-mail: user-help@spark.apache.org<mailto:user-help@spark.apache.org>






Mime
View raw message