{"action":"created","issue":{"url":"https://api.github.com/repos/TargetHolding/pyspark-cassandra/issues/70","labels_url":"https://api.github.com/repos/TargetHolding/pyspark-cassandra/issues/70/labels{/name}","comments_url":"https://api.github.com/repos/TargetHolding/pyspark-cassandra/issues/70/comments","events_url":"https://api.github.com/repos/TargetHolding/pyspark-cassandra/issues/70/events","html_url":"https://github.com/TargetHolding/pyspark-cassandra/issues/70","id":122284140,"number":70,"title":"new pyspark_cassandra not working on cluster","user":{"login":"mwicat","id":195912,"avatar_url":"https://avatars.githubusercontent.com/u/195912?v=3","gravatar_id":"","url":"https://api.github.com/users/mwicat","html_url":"https://github.com/mwicat","followers_url":"https://api.github.com/users/mwicat/followers","following_url":"https://api.github.com/users/mwicat/following{/other_user}","gists_url":"https://api.github.com/users/mwicat/gists{/gist_id}","starred_url":"https://api.github.com/users/mwicat/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mwicat/subscriptions","organizations_url":"https://api.github.com/users/mwicat/orgs","repos_url":"https://api.github.com/users/mwicat/repos","events_url":"https://api.github.com/users/mwicat/events{/privacy}","received_events_url":"https://api.github.com/users/mwicat/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2015-12-15T14:29:09Z","updated_at":"2015-12-15T14:50:29Z","closed_at":null,"body":"@frensjan:\r\n\r\npyspark_cassandra stopped working on our cluster after update to 0.2.x. It works when ran in local mode, though:\r\n\r\n```python\r\nPython 2.7.6 (default, Jun 22 2015, 17:58:13) \r\n\r\nIn [2]: from pyspark_cassandra import CassandraSparkContext\r\n\r\nIn [3]: sc = CassandraSparkContext('local')\r\n\r\nIn [4]: sc.cassandraTable(\"test\", \"user_quota\").first()\r\nOut[4]: Row(years=64, user_id=u'98')\r\n\r\nIn [5]: sc.stop()\r\n\r\nIn [6]: sc = CassandraSparkContext('spark://127.0.0.1:7077')\r\n\r\nIn [7]: sc.cassandraTable(\"test\", \"user_quota\").first()\r\nOut[7]: {u'__class__': u'pyspark_cassandra.Row'}\r\n```\r\n\r\nI took some time and discovered that PlainRowPickler which is responsible for turning spark-connector Row into pyspark_cassandra.Row is not being registered in net.razorvine.pickle.Pickler on executor nodes. Because of this pyrolite's Pickler interprets it as javabean and this results in invalid object.\r\n\r\nThe problem seems to be that Pickling.register() is called from PythonHelper initialization block which is not executed inside spark executors (which are separate processes so they don't have previously initialized Pickler in memory). When ran in local mode it works ok, because (at least I presume so) there is only one process and static variable holding picklers in net.razorvine.pickle.Pickler is just shared in memory:\r\n\r\n```java\r\nprivate static Map<Class<?>, IObjectPickler> customPicklers=new HashMap<Class<?>, IObjectPickler>();\r\n```\r\n\r\nWhen I move Pickling.register() to the place that I know is being executed inside executors, like BatchPickler.apply, then it works ok:\r\n\r\n```diff\r\ndiff --git a/src/main/scala/pyspark_util/Pickling.scala b/src/main/scala/pyspark_util/Pickling.scala\r\nindex a255d9c..2a03c66 100644\r\n--- a/src/main/scala/pyspark_util/Pickling.scala\r\n+++ b/src/main/scala/pyspark_util/Pickling.scala\r\n@@ -105,6 +105,7 @@ class BatchPickler(batchSize: Int = 1000)\r\n     with Serializable {\r\n \r\n   def apply(in: Iterator[_]): Iterator[Array[Byte]] = {\r\n+   pyspark_cassandra.Pickling.register()\r\n     val pickler = new Pickler()\r\n     in.grouped(batchSize).map { b => pickler.dumps(b.toArray) }\r\n   }\r\n```\r\n\r\nI don't know if BatchPickler.apply is the best place for this to call, but you should know that in current revision pyspark_cassandra won't run on cluster at all."},"comment":{"url":"https://api.github.com/repos/TargetHolding/pyspark-cassandra/issues/comments/164786025","html_url":"https://github.com/TargetHolding/pyspark-cassandra/issues/70#issuecomment-164786025","issue_url":"https://api.github.com/repos/TargetHolding/pyspark-cassandra/issues/70","id":164786025,"user":{"login":"frensjan","id":3985389,"avatar_url":"https://avatars.githubusercontent.com/u/3985389?v=3","gravatar_id":"","url":"https://api.github.com/users/frensjan","html_url":"https://github.com/frensjan","followers_url":"https://api.github.com/users/frensjan/followers","following_url":"https://api.github.com/users/frensjan/following{/other_user}","gists_url":"https://api.github.com/users/frensjan/gists{/gist_id}","starred_url":"https://api.github.com/users/frensjan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/frensjan/subscriptions","organizations_url":"https://api.github.com/users/frensjan/orgs","repos_url":"https://api.github.com/users/frensjan/repos","events_url":"https://api.github.com/users/frensjan/events{/privacy}","received_events_url":"https://api.github.com/users/frensjan/received_events","type":"User","site_admin":false},"created_at":"2015-12-15T14:50:29Z","updated_at":"2015-12-15T14:50:29Z","body":"@mwicat thanks! makes sense, stupid that I haven't thought about that the `PythonHelper` class isn't touched in the JVM's of slaves. In `BatchPickler` is an option. Probably in the constructor (class body) is better than on each time `apply` is invoked. I need to check if that covers every corner ..."}}