CODEWITHSUNDEEP
pythonapache-sparkpysparkapache-spark-sqlrdd
Jerry George
Jerry George

Reputation: 335

Convert and RDD to Spark Dataframe (Pyspark). This worked. But giving new error

I have an RDD:

rd.take(2)

[Row(id=0, items=['ab', 'nccd], actor='brad'),
 Row(id=1, items=['rd', 'fh'], actor='tony')]

I am trying to convert it to a spark dataframe:

df = spark.createDataFrame(rd)

This worked for me.

But now when I am trying to run it:

df.show()

This is giving me error. This was working. Please give me some insight on this

Error:

Py4JJavaError: An error occurred while calling o1264.showString.
: java.lang.IllegalStateException: SparkContext has been shutdown
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2021)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2050)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2069)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:336)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:2861)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2150)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2150)
at org.apache.spark.sql.Dataset$$anonfun$55.apply(Dataset.scala:2842)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:2841)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2150)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2363)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:241)
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:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:280)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

Upvotes: 0

Views: 1428

Answers (2)

Sivaprasanna Sethuraman
Sivaprasanna Sethuraman

Reputation: 4132

On top of what @pissall said, the following one should work:

from pyspark.sql.types import *

schema = StructType([StructField('id', IntegerType()), 
                     StructField('items', ArrayType(StringType())), 
                     StructField('actor', StringType())
                    ])
df = spark.createDataFrame(rd, schema)

Upvotes: 1

pissall
pissall

Reputation: 7399

As you might know Apache Spark is a lazy evaluator. There are Actions and Transformations that you can perform. Transformations are all called when an action is called. So when you make a show() or collect() call, all the functions you called before that will be processed. So your call to createDataFrame clearly didn't work.

Please have a read in this post which will give you an idea how to achieve your desired output : Creating a DataFrame from Row results in 'infer schema issue'

Upvotes: 1

Related Questions