[
https://issues.apache.org/jira/browse/SPARK-27613?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andres Fernandez updated SPARK-27613:
-------------------------------------
Description:
(Code included at the bottom)
The function "+create_dataframes_from_azure_responses_rdd+" receives
*table_names* (_list_ of _str_) and *responses_rdd* (rdd of tuples <_str_,
_str_, _int_, _tuple_, _azure.loganalytics.models.QueryResponse_>). It will
then go ahead and iterate over the table names to create dataframes filtering
the RDDs by the first element and valid response.
So far so good.
QueryResponse object (from azure.loganalytics package) contains, essentialy, a
list with 1 "_table_" which in turn has a "_columns_" and a "_rows_" field.
Every single response (fifth element of the tuple [4]) for the same table name
(first element of the tuple [0]) has exactly the same columns in the same order
(order is not important other thant to reference the rows data inside the same
response anyways). The types are stored in *column_types* taking the first
response as the sample.
Now to the tricky part.
I call flatMap on the *responses_rdd* with the function
+"tabularize_response_rdd+" which basically creates a Row object for every row
(_list_ of _str_) in the _QueryResponse_. I also create the schema based on a
*type_map* from azure types to spark.sql.types in order to specify it to the
subsequent createDataFrame instruction. If the result of this flatMap,
*table_tabular_rdd*, is not cached before creating the DataFrame from the Rows
RDD everything works smoothly. Nevertheless if the result of the flatMap,
*table_tabular_rdd*, is cached the before creating the DataFrame a mismatch is
evidenced between the actual key:values for the Row objects.
It would be good to point that when a Row Object is created from an unpacked
dict the code in
[[https://github.com/apache/spark/blob/480357cc6d71c682fe703611c71c1e6a36e6ce9a/python/pyspark/sql/types.py#L1374-L1375]](here)
sorts the keys; is this behaviour overriden somehow by caching?
Let me please know what I am doing wrong, is there any best practice /
documented solution I am not following? Im just a beginner when it comes to
Spark and would happily accept any suggestion. I hope I was clear enough, and I
am open to give you any additional details that might be helpful. Thank you!
(Code and error attached as well).
The error looks like if it was related to casting, but it can be seen that the
contents do not correspond to the key. *record_count* key is actually a Long
but in the Row it got somehow swapped for another key's value, in this case
'n/a'.
{code:java}
def create_dataframes_from_azure_responses_rdd(table_names: list,
responses_rdd: pyspark.rdd, verbose:bool=False) -> list:
ws_column_name = "WorkspaceId"
def tabularize_response_rdd(x: tuple):
import pyspark
tn, wsid, count, interval, response = x
ret = []
if response.tables[0].rows:
ret = [pyspark.sql.Row(**{ws_column_name:wsid, **{fi.name:r[i] for i,fi
in enumerate(response.tables[0].columns)}}) for r in response.tables[0].rows]
return ret
data_frames = {}
for tn in table_names:
if verbose: print("Filtering RDD items for {}".format(tn))
table_response_rdd = responses_rdd.filter(lambda x: x[0]==tn and
x[4]!=None).cache()
data_frames[tn] = None
if verbose: print("Checking if RDD for {} has data".format(tn))
if not table_response_rdd.isEmpty():
if verbose: print("Getting column types for {} from azure
response".format(tn))
column_types = {f.name:f.type for f in
table_response_rdd.take(1)[0][4].tables[0].columns}
column_types[ws_column_name] = "string"
if verbose: print("Generating pyspark.sql.Row RDD for {}".format(tn))
table_tabular_rdd = table_response_rdd.flatMap(tabularize_response_rdd)
#.cache() #Error with cache, no error without!
if verbose: print("Getting sample row for {}".format(tn))
row_fields = table_tabular_rdd.take(1)[0].asDict().keys()
if verbose: print("Building schema for {} from sample row and column
types".format(tn))
current_schema = StructType([StructField(f, type_map[column_types[f]](),
True) for f in row_fields])
if verbose: print("Creating dataframe for {}".format(tn))
table_df = spark.createDataFrame(table_tabular_rdd,
schema=current_schema).cache()
if verbose: print("Calculating expected count for {}".format(tn))
expected_count = table_response_rdd.map(lambda x:
(x[1],x[2])).distinct().map(lambda x: x[1]).sum()
real_count =
table_df.select("record_count").groupBy().sum().collect()[0][0]
table_response_rdd.unpersist()
#table_tabular_rdd.unpersist()
if verbose: print("Expected count {} vs Real count
{}".format(expected_count, real_count))
data_frames[tn]=table_df
else:
if verbose: print("{} table was empty!".format(tn))
return data_frames
{code}
{noformat}
Py4JJavaError Traceback (most recent call last) <command-2824384765475765> in
<module>() 1 resrdds = get_data_for_timespan_accross_laws(wss, tns, 1,
sta, container, sta_key, tid, creds, 5000, True) 2 resrdds.cache() ---->
3 dfs_raw = create_dataframes_from_azure_responses_rdd(tns, resrdds, True)
4 resrdds.unpersist() <command-2824384765475774> in
create_dataframes_from_azure_responses_rdd(table_names, responses_rdd, verbose)
37 if verbose: print("Calculating expected count for {}".format(tn))
38 expected_count = table_response_rdd.map(lambda x:
(x[1],x[2])).distinct().map(lambda x: x[1]).sum() ---> 39 real_count =
table_df.select("record_count").groupBy().sum().collect()[0][0] 40
table_response_rdd.unpersist() 41 #table_tabular_rdd.unpersist()
/databricks/spark/python/pyspark/sql/dataframe.py in collect(self) 546 #
Default path used in OSS Spark / for non-DF-ACL clusters: 547 with
SCCallSiteSync(self._sc) as css: --> 548 sock_info =
self._jdf.collectToPython() 549 return list(_load_from_socket(sock_info,
BatchedSerializer(PickleSerializer()))) 550
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in
__call__(self, *args) 1255 answer =
self.gateway_client.send_command(command) 1256 return_value =
get_return_value( -> 1257 answer, self.gateway_client,
self.target_id, self.name)
1258 1259 for temp_arg in temp_args:
/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw) 61 def
deco(*a, **kw): 62 try: ---> 63 return f(*a, **kw) 64
except py4j.protocol.Py4JJavaError as e: 65 s =
e.java_exception.toString()
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in
get_return_value(answer, gateway_client, target_id, name) 326 raise
Py4JJavaError( 327 "An error occurred while calling {0}{1}{2}.\n". --> 328
format(target_id, ".", name), value)
329 else: 330 raise Py4JError( Py4JJavaError: An error occurred while
calling o700.collectToPython. : org.apache.spark.SparkException: Job aborted
due to stage failure: Task 58 in stage 141.0 failed 4 times, most recent
failure: Lost task 58.3 in stage 141.0 (TID 76193, 10.139.64.12, executor 0):
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/databricks/spark/python/pyspark/worker.py", line 403, in main process()
File "/databricks/spark/python/pyspark/worker.py", line 398, in process
serializer.dump_stream(func(split_index, iterator), outfile) File
"/databricks/spark/python/pyspark/serializers.py", line 413, in dump_stream vs
= list(itertools.islice(iterator, batch)) File
"/databricks/spark/python/pyspark/util.py", line 99, in wrapper return f(*args,
**kwargs) File "/databricks/spark/python/pyspark/sql/session.py", line 785, in
prepare verify_func(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1389, in verify verify_value(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1370, in verify_struct
verifier(v) File "/databricks/spark/python/pyspark/sql/types.py", line 1389, in
verify verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1383, in verify_default verify_acceptable_types(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1278, in
verify_acceptable_types % (dataType, obj, type(obj)))) TypeError: field
record_count: LongType can not accept object 'n/a' in type <class 'str'> at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:490)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:626)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:609)
at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:444)
at
org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:634)
at
org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anonfun$1$$anon$1.hasNext(InMemoryRelation.scala:133)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1170)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1161)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1096) at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1161) at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:883)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:351) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:302) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at
org.apache.spark.scheduler.Task.doRunTask(Task.scala:139) at
org.apache.spark.scheduler.Task.run(Task.scala:112) at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748) Driver stacktrace: at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2100)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2088)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2087)
at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at
org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2087) at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
at scala.Option.foreach(Option.scala:257) at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1076)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2319)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2267)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2255)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) at
org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:873) at
org.apache.spark.SparkContext.runJob(SparkContext.scala:2252) at
org.apache.spark.SparkContext.runJob(SparkContext.scala:2350) at
org.apache.spark.sql.execution.collect.Collector.runSparkJobs(Collector.scala:234)
at
org.apache.spark.sql.execution.collect.Collector.collect(Collector.scala:269)
at
org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:69)
at
org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:75)
at
org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:497)
at
org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:469)
at
org.apache.spark.sql.execution.SparkPlan.executeCollectResult(SparkPlan.scala:312)
at org.apache.spark.sql.Dataset$$anonfun$51.apply(Dataset.scala:3289) at
org.apache.spark.sql.Dataset$$anonfun$51.apply(Dataset.scala:3288) at
org.apache.spark.sql.Dataset$$anonfun$55.apply(Dataset.scala:3423) at
org.apache.spark.sql.execution.SQLExecution$$anonfun$withCustomExecutionEnv$1.apply(SQLExecution.scala:99)
at
org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:228)
at
org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:85)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:158)
at
org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$withAction(Dataset.scala:3422)
at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:3288) 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:380) at
py4j.Gateway.invoke(Gateway.java:295) at
py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at
py4j.commands.CallCommand.execute(CallCommand.java:79) at
py4j.GatewayConnection.run(GatewayConnection.java:251) at
java.lang.Thread.run(Thread.java:748) Caused by:
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/databricks/spark/python/pyspark/worker.py", line 403, in main process()
File "/databricks/spark/python/pyspark/worker.py", line 398, in process
serializer.dump_stream(func(split_index, iterator), outfile) File
"/databricks/spark/python/pyspark/serializers.py", line 413, in dump_stream vs
= list(itertools.islice(iterator, batch)) File
"/databricks/spark/python/pyspark/util.py", line 99, in wrapper return f(*args,
**kwargs) File "/databricks/spark/python/pyspark/sql/session.py", line 785, in
prepare verify_func(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1389, in verify verify_value(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1370, in verify_struct
verifier(v) File "/databricks/spark/python/pyspark/sql/types.py", line 1389, in
verify verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1383, in verify_default verify_acceptable_types(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1278, in
verify_acceptable_types % (dataType, obj, type(obj)))) TypeError: field
record_count: LongType can not accept object 'n/a' in type <class 'str'> at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:490)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:626)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:609)
at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:444)
at
org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:634)
at
org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anonfun$1$$anon$1.hasNext(InMemoryRelation.scala:133)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1170)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1161)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1096) at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1161) at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:883)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:351) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:302) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at
org.apache.spark.scheduler.Task.doRunTask(Task.scala:139) at
org.apache.spark.scheduler.Task.run(Task.scala:112) at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
{noformat}
was:
The function "+create_dataframes_from_azure_responses_rdd+" receives
*table_names* (_list_ of _str_) and *responses_rdd* (rdd of tuples <_str_,
_str_, _int_, _tuple_, _azure.loganalytics.models.QueryResponse_>). It will
then go ahead and iterate over the table names to create dataframes filtering
the RDDs by the first element and valid response.
So far so good.
QueryResponse object (from azure.loganalytics package) contains, essentialy, a
list with 1 "_table_" which in turn has a "_columns_" and a "_rows_" field.
Every single response (fifth element of the tuple [4]) for the same table name
(first element of the tuple [0]) has exactly the same columns in the same order
(order is not important other thant to reference the rows data inside the same
response anyways). The types are stored in *column_types* taking the first
response as the sample.
Now to the tricky part.
I call flatMap on the *responses_rdd* with the function
+"tabularize_response_rdd+" which basically creates a Row object for every row
(_list_ of _str_) in the _QueryResponse_. I also create the schema based on a
*type_map* from azure types to spark.sql.types in order to specify it to the
subsequent createDataFrame instruction. If the result of this flatMap,
*table_tabular_rdd*, is not cached before creating the DataFrame from the Rows
RDD everything works smoothly. Nevertheless if the result of the flatMap,
*table_tabular_rdd*, is cached the before creating the DataFrame a mismatch is
evidenced between the actual key:values for the Row objects.
It would be good to point that when a Row Object is created from an unpacked
dict the code in
[[https://github.com/apache/spark/blob/480357cc6d71c682fe703611c71c1e6a36e6ce9a/python/pyspark/sql/types.py#L1374-L1375]](here)
sorts the keys; is this behaviour overriden somehow by caching?
Let me please know what I am doing wrong, is there any best practice /
documented solution I am not following? Im just a beginner when it comes to
Spark and would happily accept any suggestion. I hope I was clear enough, and I
am open to give you any additional details that might be helpful. Thank you!
(Code and error attached as well).
The error looks like if it was related to casting, but it can be seen that the
contents do not correspond to the key. *record_count* key is actually a Long
but in the Row it got somehow swapped for another key's value, in this case
'n/a'.
{code:java}
def create_dataframes_from_azure_responses_rdd(table_names: list,
responses_rdd: pyspark.rdd, verbose:bool=False) -> list:
ws_column_name = "WorkspaceId"
def tabularize_response_rdd(x: tuple):
import pyspark
tn, wsid, count, interval, response = x
ret = []
if response.tables[0].rows:
ret = [pyspark.sql.Row(**{ws_column_name:wsid, **{fi.name:r[i] for i,fi
in enumerate(response.tables[0].columns)}}) for r in response.tables[0].rows]
return ret
data_frames = {}
for tn in table_names:
if verbose: print("Filtering RDD items for {}".format(tn))
table_response_rdd = responses_rdd.filter(lambda x: x[0]==tn and
x[4]!=None).cache()
data_frames[tn] = None
if verbose: print("Checking if RDD for {} has data".format(tn))
if not table_response_rdd.isEmpty():
if verbose: print("Getting column types for {} from azure
response".format(tn))
column_types = {f.name:f.type for f in
table_response_rdd.take(1)[0][4].tables[0].columns}
column_types[ws_column_name] = "string"
if verbose: print("Generating pyspark.sql.Row RDD for {}".format(tn))
table_tabular_rdd = table_response_rdd.flatMap(tabularize_response_rdd)
#.cache() #Error with cache, no error without!
if verbose: print("Getting sample row for {}".format(tn))
row_fields = table_tabular_rdd.take(1)[0].asDict().keys()
if verbose: print("Building schema for {} from sample row and column
types".format(tn))
current_schema = StructType([StructField(f, type_map[column_types[f]](),
True) for f in row_fields])
if verbose: print("Creating dataframe for {}".format(tn))
table_df = spark.createDataFrame(table_tabular_rdd,
schema=current_schema).cache()
if verbose: print("Calculating expected count for {}".format(tn))
expected_count = table_response_rdd.map(lambda x:
(x[1],x[2])).distinct().map(lambda x: x[1]).sum()
real_count =
table_df.select("record_count").groupBy().sum().collect()[0][0]
table_response_rdd.unpersist()
#table_tabular_rdd.unpersist()
if verbose: print("Expected count {} vs Real count
{}".format(expected_count, real_count))
data_frames[tn]=table_df
else:
if verbose: print("{} table was empty!".format(tn))
return data_frames
{code}
{noformat}
Py4JJavaError Traceback (most recent call last) <command-2824384765475765> in
<module>() 1 resrdds = get_data_for_timespan_accross_laws(wss, tns, 1,
sta, container, sta_key, tid, creds, 5000, True) 2 resrdds.cache() ---->
3 dfs_raw = create_dataframes_from_azure_responses_rdd(tns, resrdds, True)
4 resrdds.unpersist() <command-2824384765475774> in
create_dataframes_from_azure_responses_rdd(table_names, responses_rdd, verbose)
37 if verbose: print("Calculating expected count for {}".format(tn))
38 expected_count = table_response_rdd.map(lambda x:
(x[1],x[2])).distinct().map(lambda x: x[1]).sum() ---> 39 real_count =
table_df.select("record_count").groupBy().sum().collect()[0][0] 40
table_response_rdd.unpersist() 41 #table_tabular_rdd.unpersist()
/databricks/spark/python/pyspark/sql/dataframe.py in collect(self) 546 #
Default path used in OSS Spark / for non-DF-ACL clusters: 547 with
SCCallSiteSync(self._sc) as css: --> 548 sock_info =
self._jdf.collectToPython() 549 return list(_load_from_socket(sock_info,
BatchedSerializer(PickleSerializer()))) 550
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in
__call__(self, *args) 1255 answer =
self.gateway_client.send_command(command) 1256 return_value =
get_return_value( -> 1257 answer, self.gateway_client,
self.target_id, self.name)
1258 1259 for temp_arg in temp_args:
/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw) 61 def
deco(*a, **kw): 62 try: ---> 63 return f(*a, **kw) 64
except py4j.protocol.Py4JJavaError as e: 65 s =
e.java_exception.toString()
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in
get_return_value(answer, gateway_client, target_id, name) 326 raise
Py4JJavaError( 327 "An error occurred while calling {0}{1}{2}.\n". --> 328
format(target_id, ".", name), value)
329 else: 330 raise Py4JError( Py4JJavaError: An error occurred while
calling o700.collectToPython. : org.apache.spark.SparkException: Job aborted
due to stage failure: Task 58 in stage 141.0 failed 4 times, most recent
failure: Lost task 58.3 in stage 141.0 (TID 76193, 10.139.64.12, executor 0):
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/databricks/spark/python/pyspark/worker.py", line 403, in main process()
File "/databricks/spark/python/pyspark/worker.py", line 398, in process
serializer.dump_stream(func(split_index, iterator), outfile) File
"/databricks/spark/python/pyspark/serializers.py", line 413, in dump_stream vs
= list(itertools.islice(iterator, batch)) File
"/databricks/spark/python/pyspark/util.py", line 99, in wrapper return f(*args,
**kwargs) File "/databricks/spark/python/pyspark/sql/session.py", line 785, in
prepare verify_func(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1389, in verify verify_value(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1370, in verify_struct
verifier(v) File "/databricks/spark/python/pyspark/sql/types.py", line 1389, in
verify verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1383, in verify_default verify_acceptable_types(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1278, in
verify_acceptable_types % (dataType, obj, type(obj)))) TypeError: field
record_count: LongType can not accept object 'n/a' in type <class 'str'> at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:490)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:626)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:609)
at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:444)
at
org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:634)
at
org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anonfun$1$$anon$1.hasNext(InMemoryRelation.scala:133)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1170)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1161)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1096) at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1161) at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:883)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:351) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:302) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at
org.apache.spark.scheduler.Task.doRunTask(Task.scala:139) at
org.apache.spark.scheduler.Task.run(Task.scala:112) at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748) Driver stacktrace: at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2100)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2088)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2087)
at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at
org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2087) at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
at scala.Option.foreach(Option.scala:257) at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1076)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2319)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2267)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2255)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) at
org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:873) at
org.apache.spark.SparkContext.runJob(SparkContext.scala:2252) at
org.apache.spark.SparkContext.runJob(SparkContext.scala:2350) at
org.apache.spark.sql.execution.collect.Collector.runSparkJobs(Collector.scala:234)
at
org.apache.spark.sql.execution.collect.Collector.collect(Collector.scala:269)
at
org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:69)
at
org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:75)
at
org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:497)
at
org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:469)
at
org.apache.spark.sql.execution.SparkPlan.executeCollectResult(SparkPlan.scala:312)
at org.apache.spark.sql.Dataset$$anonfun$51.apply(Dataset.scala:3289) at
org.apache.spark.sql.Dataset$$anonfun$51.apply(Dataset.scala:3288) at
org.apache.spark.sql.Dataset$$anonfun$55.apply(Dataset.scala:3423) at
org.apache.spark.sql.execution.SQLExecution$$anonfun$withCustomExecutionEnv$1.apply(SQLExecution.scala:99)
at
org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:228)
at
org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:85)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:158)
at
org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$withAction(Dataset.scala:3422)
at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:3288) 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:380) at
py4j.Gateway.invoke(Gateway.java:295) at
py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at
py4j.commands.CallCommand.execute(CallCommand.java:79) at
py4j.GatewayConnection.run(GatewayConnection.java:251) at
java.lang.Thread.run(Thread.java:748) Caused by:
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/databricks/spark/python/pyspark/worker.py", line 403, in main process()
File "/databricks/spark/python/pyspark/worker.py", line 398, in process
serializer.dump_stream(func(split_index, iterator), outfile) File
"/databricks/spark/python/pyspark/serializers.py", line 413, in dump_stream vs
= list(itertools.islice(iterator, batch)) File
"/databricks/spark/python/pyspark/util.py", line 99, in wrapper return f(*args,
**kwargs) File "/databricks/spark/python/pyspark/sql/session.py", line 785, in
prepare verify_func(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1389, in verify verify_value(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1370, in verify_struct
verifier(v) File "/databricks/spark/python/pyspark/sql/types.py", line 1389, in
verify verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py",
line 1383, in verify_default verify_acceptable_types(obj) File
"/databricks/spark/python/pyspark/sql/types.py", line 1278, in
verify_acceptable_types % (dataType, obj, type(obj)))) TypeError: field
record_count: LongType can not accept object 'n/a' in type <class 'str'> at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:490)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:626)
at
org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:609)
at
org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:444)
at
org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:634)
at
org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anonfun$1$$anon$1.hasNext(InMemoryRelation.scala:133)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1170)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1161)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1096) at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1161) at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:883)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:351) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:302) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at
org.apache.spark.scheduler.Task.doRunTask(Task.scala:139) at
org.apache.spark.scheduler.Task.run(Task.scala:112) at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
{noformat}
> Caching an RDD composed of Row Objects produces some kind of key recombination
> ------------------------------------------------------------------------------
>
> Key: SPARK-27613
> URL: https://issues.apache.org/jira/browse/SPARK-27613
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 2.4.0
> Reporter: Andres Fernandez
> Priority: Major
>
> (Code included at the bottom)
> The function "+create_dataframes_from_azure_responses_rdd+" receives
> *table_names* (_list_ of _str_) and *responses_rdd* (rdd of tuples <_str_,
> _str_, _int_, _tuple_, _azure.loganalytics.models.QueryResponse_>). It will
> then go ahead and iterate over the table names to create dataframes filtering
> the RDDs by the first element and valid response.
> So far so good.
> QueryResponse object (from azure.loganalytics package) contains, essentialy,
> a list with 1 "_table_" which in turn has a "_columns_" and a "_rows_" field.
> Every single response (fifth element of the tuple [4]) for the same table
> name (first element of the tuple [0]) has exactly the same columns in the
> same order (order is not important other thant to reference the rows data
> inside the same response anyways). The types are stored in *column_types*
> taking the first response as the sample.
> Now to the tricky part.
> I call flatMap on the *responses_rdd* with the function
> +"tabularize_response_rdd+" which basically creates a Row object for every
> row (_list_ of _str_) in the _QueryResponse_. I also create the schema based
> on a *type_map* from azure types to spark.sql.types in order to specify it to
> the subsequent createDataFrame instruction. If the result of this flatMap,
> *table_tabular_rdd*, is not cached before creating the DataFrame from the
> Rows RDD everything works smoothly. Nevertheless if the result of the
> flatMap, *table_tabular_rdd*, is cached the before creating the DataFrame a
> mismatch is evidenced between the actual key:values for the Row objects.
> It would be good to point that when a Row Object is created from an unpacked
> dict the code in
> [[https://github.com/apache/spark/blob/480357cc6d71c682fe703611c71c1e6a36e6ce9a/python/pyspark/sql/types.py#L1374-L1375]](here)
> sorts the keys; is this behaviour overriden somehow by caching?
> Let me please know what I am doing wrong, is there any best practice /
> documented solution I am not following? Im just a beginner when it comes to
> Spark and would happily accept any suggestion. I hope I was clear enough, and
> I am open to give you any additional details that might be helpful. Thank
> you! (Code and error attached as well).
> The error looks like if it was related to casting, but it can be seen that
> the contents do not correspond to the key. *record_count* key is actually a
> Long but in the Row it got somehow swapped for another key's value, in this
> case 'n/a'.
> {code:java}
> def create_dataframes_from_azure_responses_rdd(table_names: list,
> responses_rdd: pyspark.rdd, verbose:bool=False) -> list:
> ws_column_name = "WorkspaceId"
> def tabularize_response_rdd(x: tuple):
> import pyspark
> tn, wsid, count, interval, response = x
> ret = []
> if response.tables[0].rows:
> ret = [pyspark.sql.Row(**{ws_column_name:wsid, **{fi.name:r[i] for i,fi
> in enumerate(response.tables[0].columns)}}) for r in response.tables[0].rows]
> return ret
> data_frames = {}
> for tn in table_names:
> if verbose: print("Filtering RDD items for {}".format(tn))
> table_response_rdd = responses_rdd.filter(lambda x: x[0]==tn and
> x[4]!=None).cache()
>
> data_frames[tn] = None
> if verbose: print("Checking if RDD for {} has data".format(tn))
> if not table_response_rdd.isEmpty():
> if verbose: print("Getting column types for {} from azure
> response".format(tn))
> column_types = {f.name:f.type for f in
> table_response_rdd.take(1)[0][4].tables[0].columns}
> column_types[ws_column_name] = "string"
> if verbose: print("Generating pyspark.sql.Row RDD for {}".format(tn))
> table_tabular_rdd = table_response_rdd.flatMap(tabularize_response_rdd)
> #.cache() #Error with cache, no error without!
> if verbose: print("Getting sample row for {}".format(tn))
> row_fields = table_tabular_rdd.take(1)[0].asDict().keys()
> if verbose: print("Building schema for {} from sample row and column
> types".format(tn))
> current_schema = StructType([StructField(f,
> type_map[column_types[f]](), True) for f in row_fields])
> if verbose: print("Creating dataframe for {}".format(tn))
> table_df = spark.createDataFrame(table_tabular_rdd,
> schema=current_schema).cache()
> if verbose: print("Calculating expected count for {}".format(tn))
> expected_count = table_response_rdd.map(lambda x:
> (x[1],x[2])).distinct().map(lambda x: x[1]).sum()
> real_count =
> table_df.select("record_count").groupBy().sum().collect()[0][0]
> table_response_rdd.unpersist()
> #table_tabular_rdd.unpersist()
> if verbose: print("Expected count {} vs Real count
> {}".format(expected_count, real_count))
> data_frames[tn]=table_df
> else:
> if verbose: print("{} table was empty!".format(tn))
> return data_frames
> {code}
> {noformat}
> Py4JJavaError Traceback (most recent call last) <command-2824384765475765> in
> <module>() 1 resrdds = get_data_for_timespan_accross_laws(wss, tns, 1,
> sta, container, sta_key, tid, creds, 5000, True) 2 resrdds.cache()
> ----> 3 dfs_raw = create_dataframes_from_azure_responses_rdd(tns, resrdds,
> True) 4 resrdds.unpersist() <command-2824384765475774> in
> create_dataframes_from_azure_responses_rdd(table_names, responses_rdd,
> verbose) 37 if verbose: print("Calculating expected count for
> {}".format(tn)) 38 expected_count = table_response_rdd.map(lambda x:
> (x[1],x[2])).distinct().map(lambda x: x[1]).sum() ---> 39 real_count =
> table_df.select("record_count").groupBy().sum().collect()[0][0] 40
> table_response_rdd.unpersist() 41 #table_tabular_rdd.unpersist()
> /databricks/spark/python/pyspark/sql/dataframe.py in collect(self) 546 #
> Default path used in OSS Spark / for non-DF-ACL clusters: 547 with
> SCCallSiteSync(self._sc) as css: --> 548 sock_info =
> self._jdf.collectToPython() 549 return list(_load_from_socket(sock_info,
> BatchedSerializer(PickleSerializer()))) 550
> /databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in
> __call__(self, *args) 1255 answer =
> self.gateway_client.send_command(command) 1256 return_value =
> get_return_value( -> 1257 answer, self.gateway_client,
> self.target_id, self.name)
> 1258 1259 for temp_arg in temp_args:
> /databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw) 61 def
> deco(*a, **kw): 62 try: ---> 63 return f(*a, **kw) 64
> except py4j.protocol.Py4JJavaError as e: 65 s =
> e.java_exception.toString()
> /databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in
> get_return_value(answer, gateway_client, target_id, name) 326 raise
> Py4JJavaError( 327 "An error occurred while calling {0}{1}{2}.\n". -->
> 328 format(target_id, ".", name), value)
> 329 else: 330 raise Py4JError( Py4JJavaError: An error occurred while
> calling o700.collectToPython. : org.apache.spark.SparkException: Job aborted
> due to stage failure: Task 58 in stage 141.0 failed 4 times, most recent
> failure: Lost task 58.3 in stage 141.0 (TID 76193, 10.139.64.12, executor 0):
> org.apache.spark.api.python.PythonException: Traceback (most recent call
> last): File "/databricks/spark/python/pyspark/worker.py", line 403, in main
> process() File "/databricks/spark/python/pyspark/worker.py", line 398, in
> process serializer.dump_stream(func(split_index, iterator), outfile) File
> "/databricks/spark/python/pyspark/serializers.py", line 413, in dump_stream
> vs = list(itertools.islice(iterator, batch)) File
> "/databricks/spark/python/pyspark/util.py", line 99, in wrapper return
> f(*args, **kwargs) File "/databricks/spark/python/pyspark/sql/session.py",
> line 785, in prepare verify_func(obj) File
> "/databricks/spark/python/pyspark/sql/types.py", line 1389, in verify
> verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py", line
> 1370, in verify_struct verifier(v) File
> "/databricks/spark/python/pyspark/sql/types.py", line 1389, in verify
> verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py", line
> 1383, in verify_default verify_acceptable_types(obj) File
> "/databricks/spark/python/pyspark/sql/types.py", line 1278, in
> verify_acceptable_types % (dataType, obj, type(obj)))) TypeError: field
> record_count: LongType can not accept object 'n/a' in type <class 'str'> at
> org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:490)
> at
> org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:626)
> at
> org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:609)
> at
> org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:444)
> at
> org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
> at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) at
> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
> Source) at
> org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
> at
> org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:634)
> at
> org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anonfun$1$$anon$1.hasNext(InMemoryRelation.scala:133)
> at
> org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
> at
> org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
> at
> org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1170)
> at
> org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1161)
> at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1096) at
> org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1161)
> at
> org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:883)
> at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:351) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:302) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at
> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at
> org.apache.spark.scheduler.Task.doRunTask(Task.scala:139) at
> org.apache.spark.scheduler.Task.run(Task.scala:112) at
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
> at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432) at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503) at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748) Driver stacktrace: at
> org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2100)
> at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2088)
> at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2087)
> at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at
> org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2087)
> at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
> at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
> at scala.Option.foreach(Option.scala:257) at
> org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1076)
> at
> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2319)
> at
> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2267)
> at
> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2255)
> at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) at
> org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:873) at
> org.apache.spark.SparkContext.runJob(SparkContext.scala:2252) at
> org.apache.spark.SparkContext.runJob(SparkContext.scala:2350) at
> org.apache.spark.sql.execution.collect.Collector.runSparkJobs(Collector.scala:234)
> at
> org.apache.spark.sql.execution.collect.Collector.collect(Collector.scala:269)
> at
> org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:69)
> at
> org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:75)
> at
> org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:497)
> at
> org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:469)
> at
> org.apache.spark.sql.execution.SparkPlan.executeCollectResult(SparkPlan.scala:312)
> at org.apache.spark.sql.Dataset$$anonfun$51.apply(Dataset.scala:3289) at
> org.apache.spark.sql.Dataset$$anonfun$51.apply(Dataset.scala:3288) at
> org.apache.spark.sql.Dataset$$anonfun$55.apply(Dataset.scala:3423) at
> org.apache.spark.sql.execution.SQLExecution$$anonfun$withCustomExecutionEnv$1.apply(SQLExecution.scala:99)
> at
> org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:228)
> at
> org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:85)
> at
> org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:158)
> at
> org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$withAction(Dataset.scala:3422)
> at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:3288) 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:380) at
> py4j.Gateway.invoke(Gateway.java:295) at
> py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at
> py4j.commands.CallCommand.execute(CallCommand.java:79) at
> py4j.GatewayConnection.run(GatewayConnection.java:251) at
> java.lang.Thread.run(Thread.java:748) Caused by:
> org.apache.spark.api.python.PythonException: Traceback (most recent call
> last): File "/databricks/spark/python/pyspark/worker.py", line 403, in main
> process() File "/databricks/spark/python/pyspark/worker.py", line 398, in
> process serializer.dump_stream(func(split_index, iterator), outfile) File
> "/databricks/spark/python/pyspark/serializers.py", line 413, in dump_stream
> vs = list(itertools.islice(iterator, batch)) File
> "/databricks/spark/python/pyspark/util.py", line 99, in wrapper return
> f(*args, **kwargs) File "/databricks/spark/python/pyspark/sql/session.py",
> line 785, in prepare verify_func(obj) File
> "/databricks/spark/python/pyspark/sql/types.py", line 1389, in verify
> verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py", line
> 1370, in verify_struct verifier(v) File
> "/databricks/spark/python/pyspark/sql/types.py", line 1389, in verify
> verify_value(obj) File "/databricks/spark/python/pyspark/sql/types.py", line
> 1383, in verify_default verify_acceptable_types(obj) File
> "/databricks/spark/python/pyspark/sql/types.py", line 1278, in
> verify_acceptable_types % (dataType, obj, type(obj)))) TypeError: field
> record_count: LongType can not accept object 'n/a' in type <class 'str'> at
> org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:490)
> at
> org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:626)
> at
> org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:609)
> at
> org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:444)
> at
> org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
> at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439) at
> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
> Source) at
> org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
> at
> org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:634)
> at
> org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anonfun$1$$anon$1.hasNext(InMemoryRelation.scala:133)
> at
> org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
> at
> org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
> at
> org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1170)
> at
> org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1161)
> at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1096) at
> org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1161)
> at
> org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:883)
> at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:351) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:302) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60) at
> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:340) at
> org.apache.spark.rdd.RDD.iterator(RDD.scala:304) at
> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at
> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at
> org.apache.spark.scheduler.Task.doRunTask(Task.scala:139) at
> org.apache.spark.scheduler.Task.run(Task.scala:112) at
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
> at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432) at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503) at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> ... 1 more
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
