You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by YaoPau <jo...@gmail.com> on 2015/12/04 00:28:55 UTC
sparkavro for PySpark 1.3
How can I read from and write to Avro using PySpark in 1.3?
I can only find the 1.4 documentation
<https://github.com/databricks/spark-avro/blob/master/README.md> , which
uses a sqlContext.read method that isn't available to me in 1.3.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/sparkavro-for-PySpark-1-3-tp25561.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: sparkavro for PySpark 1.3
Posted by ayan guha <gu...@gmail.com>.
Hi
You can use newAPIHadoopFile and use AvroInputFormat.
On Sun, Dec 6, 2015 at 4:59 AM, YaoPau <jo...@gmail.com> wrote:
> Here's what I'm currently trying:
>
> --------------------------------------
> I'm including --packages com.databricks:spark-avro_2.10:1.0.0 in my pyspark
> call. This seems to work:
>
> Ivy Default Cache set to: /home/jrgregg/.ivy2/cache
> The jars for the packages stored in: /home/jrgregg/.ivy2/jars
> :: loading settings :: url =
>
> jar:file:/opt/cloudera/parcels/CDH-5.4.4-1.cdh5.4.4.p894.568/jars/spark-assembly-1.3.0-cdh5.4.4-hadoop2.6.0-cdh5.4.4.jar!/org/apache/ivy/core/settings/ivysettings.xml
> com.databricks#spark-avro_2.10 added as a dependency
> :: resolving dependencies :: org.apache.spark#spark-submit-parent;1.0
> confs: [default]
> found com.databricks#spark-avro_2.10;1.0.0 in central
> found org.apache.avro#avro;1.7.6 in central
> found org.codehaus.jackson#jackson-core-asl;1.9.13 in central
> found org.codehaus.jackson#jackson-mapper-asl;1.9.13 in central
> found com.thoughtworks.paranamer#paranamer;2.3 in central
> found org.xerial.snappy#snappy-java;1.0.5 in central
> found org.apache.commons#commons-compress;1.4.1 in central
> found org.tukaani#xz;1.0 in central
> found org.slf4j#slf4j-api;1.6.4 in central
> :: resolution report :: resolve 629ms :: artifacts dl 22ms
> :: modules in use:
> com.databricks#spark-avro_2.10;1.0.0 from central in [default]
> com.thoughtworks.paranamer#paranamer;2.3 from central in [default]
> org.apache.avro#avro;1.7.6 from central in [default]
> org.apache.commons#commons-compress;1.4.1 from central in [default]
> org.codehaus.jackson#jackson-core-asl;1.9.13 from central in
> [default]
> org.codehaus.jackson#jackson-mapper-asl;1.9.13 from central in
> [default]
> org.slf4j#slf4j-api;1.6.4 from central in [default]
> org.tukaani#xz;1.0 from central in [default]
> org.xerial.snappy#snappy-java;1.0.5 from central in [default]
>
> ---------------------------------------------------------------------
> | | modules || artifacts
> |
> | conf | number| search|dwnlded|evicted||
> number|dwnlded|
>
> ---------------------------------------------------------------------
> | default | 9 | 0 | 0 | 0 || 9 | 0
> |
>
> ---------------------------------------------------------------------
> :: retrieving :: org.apache.spark#spark-submit-parent
> confs: [default]
> 0 artifacts copied, 9 already retrieved (0kB/12ms)
>
> --------------------------------------
>
> Then in my code I have:
>
> df.repartition(partitions).save(save_avro, "com.databricks.spark.avro")
>
> This results in:
>
> ---------------------------------------------------------------------------
> Py4JJavaError Traceback (most recent call last)
> <ipython-input-33-8a02829528e8> in <module>()
> ----> 1 schemaBirf.repartition(partitions).save(save_avro,
> "com.databricks.spark.avro")
>
> /opt/cloudera/parcels/CDH/lib/spark/python/pyspark/sql/dataframe.py in
> save(self, path, source, mode, **options)
> 215 joptions = MapConverter().convert(options,
> 216
> self._sc._gateway._gateway_client)
> --> 217 self._jdf.save(source, jmode, joptions)
> 218
> 219 @property
>
>
> /opt/cloudera/parcels/CDH/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py
> in __call__(self, *args)
> 536 answer = self.gateway_client.send_command(command)
> 537 return_value = get_return_value(answer,
> self.gateway_client,
> --> 538 self.target_id, self.name)
> 539
> 540 for temp_arg in temp_args:
>
>
> /opt/cloudera/parcels/CDH/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py
> in get_return_value(answer, gateway_client, target_id, name)
> 298 raise Py4JJavaError(
> 299 'An error occurred while calling {0}{1}{2}.\n'.
> --> 300 format(target_id, '.', name), value)
> 301 else:
> 302 raise Py4JError(
>
> Py4JJavaError: An error occurred while calling o127.save.
> : java.lang.NoClassDefFoundError:
> org/apache/spark/sql/sources/HadoopFsRelationProvider
> at java.lang.ClassLoader.defineClass1(Native Method)
> at java.lang.ClassLoader.defineClass(ClassLoader.java:800)
> at
> java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
> at java.net.URLClassLoader.defineClass(URLClassLoader.java:449)
> at java.net.URLClassLoader.access$100(URLClassLoader.java:71)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> at
>
> org.apache.spark.sql.sources.ResolvedDataSource$.lookupDataSource(ddl.scala:265)
> at
> org.apache.spark.sql.sources.ResolvedDataSource$.apply(ddl.scala:305)
> at org.apache.spark.sql.DataFrame.save(DataFrame.scala:1123)
> at org.apache.spark.sql.DataFrame.save(DataFrame.scala:1108)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
>
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> at
>
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
> at
> py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
> at py4j.Gateway.invoke(Gateway.java:259)
> at
> py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
> at py4j.commands.CallCommand.execute(CallCommand.java:79)
> at py4j.GatewayConnection.run(GatewayConnection.java:207)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.ClassNotFoundException:
> org.apache.spark.sql.sources.HadoopFsRelationProvider
> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> ... 26 more
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/sparkavro-for-PySpark-1-3-tp25561p25574.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>
--
Best Regards,
Ayan Guha
Re: sparkavro for PySpark 1.3
Posted by YaoPau <jo...@gmail.com>.
Here's what I'm currently trying:
--------------------------------------
I'm including --packages com.databricks:spark-avro_2.10:1.0.0 in my pyspark
call. This seems to work:
Ivy Default Cache set to: /home/jrgregg/.ivy2/cache
The jars for the packages stored in: /home/jrgregg/.ivy2/jars
:: loading settings :: url =
jar:file:/opt/cloudera/parcels/CDH-5.4.4-1.cdh5.4.4.p894.568/jars/spark-assembly-1.3.0-cdh5.4.4-hadoop2.6.0-cdh5.4.4.jar!/org/apache/ivy/core/settings/ivysettings.xml
com.databricks#spark-avro_2.10 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent;1.0
confs: [default]
found com.databricks#spark-avro_2.10;1.0.0 in central
found org.apache.avro#avro;1.7.6 in central
found org.codehaus.jackson#jackson-core-asl;1.9.13 in central
found org.codehaus.jackson#jackson-mapper-asl;1.9.13 in central
found com.thoughtworks.paranamer#paranamer;2.3 in central
found org.xerial.snappy#snappy-java;1.0.5 in central
found org.apache.commons#commons-compress;1.4.1 in central
found org.tukaani#xz;1.0 in central
found org.slf4j#slf4j-api;1.6.4 in central
:: resolution report :: resolve 629ms :: artifacts dl 22ms
:: modules in use:
com.databricks#spark-avro_2.10;1.0.0 from central in [default]
com.thoughtworks.paranamer#paranamer;2.3 from central in [default]
org.apache.avro#avro;1.7.6 from central in [default]
org.apache.commons#commons-compress;1.4.1 from central in [default]
org.codehaus.jackson#jackson-core-asl;1.9.13 from central in [default]
org.codehaus.jackson#jackson-mapper-asl;1.9.13 from central in [default]
org.slf4j#slf4j-api;1.6.4 from central in [default]
org.tukaani#xz;1.0 from central in [default]
org.xerial.snappy#snappy-java;1.0.5 from central in [default]
---------------------------------------------------------------------
| | modules || artifacts |
| conf | number| search|dwnlded|evicted|| number|dwnlded|
---------------------------------------------------------------------
| default | 9 | 0 | 0 | 0 || 9 | 0 |
---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent
confs: [default]
0 artifacts copied, 9 already retrieved (0kB/12ms)
--------------------------------------
Then in my code I have:
df.repartition(partitions).save(save_avro, "com.databricks.spark.avro")
This results in:
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-33-8a02829528e8> in <module>()
----> 1 schemaBirf.repartition(partitions).save(save_avro,
"com.databricks.spark.avro")
/opt/cloudera/parcels/CDH/lib/spark/python/pyspark/sql/dataframe.py in
save(self, path, source, mode, **options)
215 joptions = MapConverter().convert(options,
216
self._sc._gateway._gateway_client)
--> 217 self._jdf.save(source, jmode, joptions)
218
219 @property
/opt/cloudera/parcels/CDH/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py
in __call__(self, *args)
536 answer = self.gateway_client.send_command(command)
537 return_value = get_return_value(answer, self.gateway_client,
--> 538 self.target_id, self.name)
539
540 for temp_arg in temp_args:
/opt/cloudera/parcels/CDH/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py
in get_return_value(answer, gateway_client, target_id, name)
298 raise Py4JJavaError(
299 'An error occurred while calling {0}{1}{2}.\n'.
--> 300 format(target_id, '.', name), value)
301 else:
302 raise Py4JError(
Py4JJavaError: An error occurred while calling o127.save.
: java.lang.NoClassDefFoundError:
org/apache/spark/sql/sources/HadoopFsRelationProvider
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(ClassLoader.java:800)
at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
at java.net.URLClassLoader.defineClass(URLClassLoader.java:449)
at java.net.URLClassLoader.access$100(URLClassLoader.java:71)
at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
at
org.apache.spark.sql.sources.ResolvedDataSource$.lookupDataSource(ddl.scala:265)
at org.apache.spark.sql.sources.ResolvedDataSource$.apply(ddl.scala:305)
at org.apache.spark.sql.DataFrame.save(DataFrame.scala:1123)
at org.apache.spark.sql.DataFrame.save(DataFrame.scala:1108)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException:
org.apache.spark.sql.sources.HadoopFsRelationProvider
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
... 26 more
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/sparkavro-for-PySpark-1-3-tp25561p25574.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org