pyspark json not working
Question: I am trying to parse [json
data](http://%7B%20%20%20%22hash%22:%220000000000000000059134ebb840559241e8e2799f3ebdff56723efecfd6567a%22,%20%20%20%22branch%22:%22main%22,%20%20%20%22previous_block_hash%22:%220000000000000000010d1517398ca2c64b055aa00ce04bcc36a0fc66fc12e76a%22,%20%20%20%22next_blocks%22:\[%20%20%20%20%20%7B%20%20%20%20%20%20%20%22hash%22:%22000000000000000002662fadc11e999216cc6add1568a6af09f2dc84a1c500b9%22,%20%20%20%20%20%20%20%22branch%22:%22main%22,%20%20%20%20%20%20%20%22height%22:395546%20%20%20%20%20%7D%20%20%20\],%20%20%20%22height%22:395545,%20%20%20%22confirmations%22:876,%20%20%20%22merkle_root%22:%228cf3eea32f692e5ebc9c25bb912ab3aff43c02761609d52cdd48afc5a05918fb%22,%20%20%20%22time%22:%222016-01-29T02:54:45Z%22,%20%20%20%22created_at%22:%222016-01-29T02:55:18Z%22,%20%20%20%22nonce%22:3284789611,%20%20%20%22bits%22:403253488,%20%20%20%22difficulty%22:120033340651.23708,%20%20%20%22reward%22:2500000000,%20%20%20%22fees%22:481000,%20%20%20%22total_out%22:94587603763,%20%20%20%22size%22:52543,%20%20%20%22transactions_count%22:28,%20%20%20%22version%22:4,%20%20%20%22transaction_hashes%22:\[%20%20%20%20%20%22b3df3d5fedadd07a46753af556c336c41e038a9aec7ddd9921ad249828fd6d66%22,%20%20%20%20%20%224ada431255d104c1c76ef56bdef4186ea89793223133e535383ff39d5a322910%22,%20%20%20%20%20%228598c3c2f9c0e319123df71b66558ea09d07a832817afcddbea050b279ec641d%22,%20%20%20%20%20%227ba6dd7775102b071727f3839cff2ad455971d6d17f93602daab79797c32f23a%22,%20%20%20%20%20%22aaad0db749d431806048073c2f96e3c49e41d87f46f0aadc505596c06fa63faf%22,%20%20%20%20%20%221ef914db9dfcf97fb6d0954506a1a360bd8eabc07a7bd32ea9d0723e8802d082%22,%20%20%20%20%20%2262d4eaac2cbd20182c53992bf69af4a1a0244f783317e141b5680120ac3c2d98%22,%20%20%20%20%20%2231f63a7f5782d324da4eb6dd716bb36723f02fc53b3cb2c900ab3f4c2df105b2%22,%20%20%20%20%20%224a1fc06ac9c39068127dca5dc042e05d58ed2bd6a786a894351392b9cefa30b3%22,%20%20%20%20%20%2231ce58b46297f0998e35be1fd93e30caa85b608d04e9565138429870a39ad2ba%22,%20%20%20%20%20%222fd5e015f3d0f2f06e9d31bfecdfa2308fa33109645ffbe4e477225efa6b4286%22,%20%20%20%20%20%2293aa6d00e82a7831909697099c3241dd7dd324ec0a93e29976bdfbbd80fcc551%22,%20%20%20%20%20%2203942d53df85cc13d8183f541f127c1e796b8c04d163fa4ada86da89f1563272%22,%20%20%20%20%20%224f6cf7d83597e7db7f5ae67573a03aba3c8e8cec38f9b61cd1e3a060827e0aac%22,%20%20%20%20%20%22c885cf7aebdbae2af3d26061de45b768de81a2720b54e573fe802a0e81056ab0%22,%20%20%20%20%20%22ea364361d8506f53934241482fefe387b6cb851a5c7058cbe88cd16d1442fba6%22,%20%20%20%20%20%22a8799fbd00ed00cb9bb95316b73e56aefb5548d2abf7d4e789c7a71d5b91fd19%22,%20%20%20%20%20%2292891df395623487dbc2c92adc418394c0c732389d4b02455d471c8fcffab34b%22,%20%20%20%20%20%2250c0cca90792e8ac141a13bdae1400aa6b36b7760d4eb453eea1b0ad40b390b3%22,%20%20%20%20%20%22557f6dda4e4388f0d94b491599aab03796f778a9612a5167b1c5435f9db4cf7c%22,%20%20%20%20%20%22dfd97faa218658fe0a69e3948074459c8f37194a8cfaa28f0419b23cc11f0599%22,%20%20%20%20%20%22351d694803761f1193ef61dcc6678f570b22aa540d5c0f0b3a0a7cc43e203c6d%22,%20%20%20%20%20%221228976f4be8e72ba7fad1e0e1b57453dc2d416865716bc1a6abadcb6c949ab6%22,%20%20%20%20%20%22022c5f6d8ea7b3ca48fabfb08bb0ebeee59acd409898aa5d3bc0fd35e49bf12b%22,%20%20%20%20%20%2274a42e5583f29937a3e66d6b11632868f9d02a9060ca809122d7e77cbe93401d%22,%20%20%20%20%20%22bd7fecc6e4e8212c54deba8845bbf4f818cddb77d9ddc02b5e7029614999cdf5%22,%20%20%20%20%20%224487bc63459c4308ef87c335798f2af681e350c1bb9699fcea8cf862326227c2%22,%20%20%20%20%20%227f35e654a7e5d8932968ca3c5de86b6f3bff0180022c3c25f9acbac81ef7c539%22%20%20%20\]%20%7D)
in Pyspark's map function. I am interested in the extracting the field
"fees":481000 from [json data](https://bitcoin.toshi.io/api/v0/blocks/395545)
on line #21. If i do this in plain python (that is without pyspark), i can do
this with the following and it works!!!

    
    
    import json
    f=open("block_395545.json")
    lines = f.read()
    json_data = json.loads(lines)
    fee_data = json_data["fees"]
    print fee_data
    

But if i put this inside a map function as follows, it does not work :(

    
    
    function get_tx_fee(line):
         json_data = json.loads(line)
         fee_data = json_data["fees"]
         print fee_data
         return fee_data
    
    lines_rdd = sc.textFile("file:///block_395545.json")
    tx_fee = lines_rdd.map(get_tx_fee) 
    

Any idea how to fix this?

Here is the code after incorporating @zero323 suggestion

    
    
    from pyspark import SparkConf, SparkContext
    import json
    conf = SparkConf().setMaster("local").setAppName("bitcoin_TransactionFee_calcultor")
    sc = SparkContext(conf=conf)
    
    
    content_rdd = (sc.wholeTextFiles("file:///home/ubuntu/unix_practice/spark-example/block_json_395545.txt")
                    .map(lambda kv: kv[0])
                    .map(json.loads)
                  )
    content_rdd.take(10)
    

and here is are the ERRORS after running spark-submit

    
    
    Traceback (most recent call last):
      File "/home/ubuntu/unix_practice/spark-example/bctxfee_text.py", line 48, in <module>
        content_rdd.take(10)
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1299, in take
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/context.py", line 916, in runJob
      File "/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 36, in deco
      File "/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
    py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
    : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
        process()
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
        serializer.dump_stream(func(split_index, iterator), outfile)
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
        vs = list(itertools.islice(iterator, batch))
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1295, in takeUpToNumLeft
      File "/usr/lib/python2.7/json/__init__.py", line 338, in loads
        return _default_decoder.decode(s)
      File "/usr/lib/python2.7/json/decoder.py", line 366, in decode
        obj, end = self.raw_decode(s, idx=_w(s, 0).end())
      File "/usr/lib/python2.7/json/decoder.py", line 384, in raw_decode
        raise ValueError("No JSON object could be decoded")
    ValueError: No JSON object could be decoded
    
        at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
        at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)
        at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
        at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
        at org.apache.spark.scheduler.Task.run(Task.scala:88)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
    
    Driver stacktrace:
        at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
        at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)
        at scala.Option.foreach(Option.scala:236)
        at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)
        at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
        at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1824)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1837)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1850)
        at org.apache.spark.api.python.PythonRDD$.runJob(PythonRDD.scala:393)
        at org.apache.spark.api.python.PythonRDD.runJob(PythonRDD.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        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: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
        process()
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
        serializer.dump_stream(func(split_index, iterator), outfile)
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
        vs = list(itertools.islice(iterator, batch))
      File "/usr/local/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1295, in takeUpToNumLeft
      File "/usr/lib/python2.7/json/__init__.py", line 338, in loads
        return _default_decoder.decode(s)
      File "/usr/lib/python2.7/json/decoder.py", line 366, in decode
        obj, end = self.raw_decode(s, idx=_w(s, 0).end())
      File "/usr/lib/python2.7/json/decoder.py", line 384, in raw_decode
        raise ValueError("No JSON object could be decoded")
    ValueError: No JSON object could be decoded
    
        at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
        at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)
        at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
        at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
        at org.apache.spark.scheduler.Task.run(Task.scala:88)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        ... 1 more
    
    16/02/03 22:42:42 INFO SparkContext: Invoking stop() from shutdown hook
    16/02/03 22:42:42 INFO SparkUI: Stopped Spark web UI at http://ec2-52-72-36-43.compute-1.amazonaws.com:4040
    16/02/03 22:42:42 INFO DAGScheduler: Stopping DAGScheduler
    16/02/03 22:42:42 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
    16/02/03 22:42:42 INFO MemoryStore: MemoryStore cleared
    16/02/03 22:42:42 INFO BlockManager: BlockManager stopped
    16/02/03 22:42:42 INFO BlockManagerMaster: BlockManagerMaster stopped
    16/02/03 22:42:42 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
    16/02/03 22:42:42 INFO SparkContext: Successfully stopped SparkContext
    16/02/03 22:42:42 INFO ShutdownHookManager: Shutdown hook called
    16/02/03 22:42:42 INFO ShutdownHookManager: Deleting directory /tmp/spark-8c82aefa-b0f4-4365-b7a9-cdb7468aed68/pyspark-47daf835-b869-48cc-9c3d-b836f2ff26b8
    16/02/03 22:42:42 INFO ShutdownHookManager: Deleting directory /tmp/spark-8c82aefa-b0f4-4365-b7a9-cdb7468aed68
    


Answer: `SparkContex.textFile` splits data by line so it won't work with multiline
JSON. If files are relatively small you can try
[`SparkContex.wholeTextFiles`](https://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.SparkContext.wholeTextFiles)
instead:

    
    
    (sc.wholeTextFiles("file:///block_395545.json")
        .map(lambda kv: kv[1])
        .map(json.loads))
    

If not you can try to leverage document structure to create custom Hadoop
`InputFormat`, [use custom
delimiter](http://stackoverflow.com/a/31836069/1560062) or try to fix data
after loading using methods which can access multiple records like
`mapPartitions` / `mapPartitionsWithIndex`.

