# How to cast incorrectly detected schema. Pyspark-ES

**URL:** <https://discuss.elastic.co/t/how-to-cast-incorrectly-detected-schema-pyspark-es/123467>\
**Category:** Elasticsearch\
**Tags:** es-hadoop\
**Created:** [March 12, 2018, 2:20am UTC](https://discuss.elastic.co/t/how-to-cast-incorrectly-detected-schema-pyspark-es/123467 "2018-03-12T02:20:45Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Pramod\_Sripada](https://avatars.discourse-cdn.com/v4/letter/p/85e7bf/32.png) [@Pramod\_Sripada](https://discuss.elastic.co/u/Pramod_Sripada)\
**Post date:** [March 12, 2018, 2:20am UTC](https://discuss.elastic.co/t/how-to-cast-incorrectly-detected-schema-pyspark-es/123467/1 "2018-03-12T02:20:46Z")

</div>

Hi,

I am reading in geo point data from ElasticSearch index using Pyspark. I am creating my DataFrame using the following command.

`us_df = spark.read.format('es').option('es.query', us_q).option('es.read.field.as.array.include', 'extra_tags').load('index').select('centroid.lat', 'centroid.lon')`

When I print the Schema of the DataFrame I get the schema as

```
root
 |-- lat: double (nullable = true)
 |-- lon: double (nullable = true)

```

But when I try to get the first 10 records

`us_df.take(10)`

I get an error

```
18/03/11 21:10:38 ERROR TaskSetManager: Task 0 in stage 6.0 failed 1 times; aborting job
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/ksripada/spark-2.1.1-bin-hadoop2.7/python/pyspark/sql/dataframe.py", line 429, in take
    return self.limit(num).collect()
  File "/Users/ksripada/spark-2.1.1-bin-hadoop2.7/python/pyspark/sql/dataframe.py", line 391, in collect
    port = self._jdf.collectToPython()
  File "/Users/ksripada/spark-2.1.1-bin-hadoop2.7/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1133, in __call__
  File "/Users/ksripada/spark-2.1.1-bin-hadoop2.7/python/pyspark/sql/utils.py", line 63, in deco
    return f(*a, **kw)
  File "/Users/ksripada/spark-2.1.1-bin-hadoop2.7/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py", line 319, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o260.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 6.0 failed 1 times, most recent failure: Lost task 0.0 in stage 6.0 (TID 9, localhost, executor driver): java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Double
	at scala.runtime.BoxesRunTime.unboxToDouble(BoxesRunTime.java:114)
	at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow$class.getDouble(rows.scala:44)
	at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.getDouble(rows.scala:194)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:231)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:225)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
	at org.apache.spark.scheduler.Task.run(Task.scala:99)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
	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)

```

I have seen the trace and the latitude and longitude is being returned as Strings from ES

`{"_index":"places","_type":"us_place","_id":"548776421","_score":null,"_source":{"centroid":{"lon":"-87.739957","lat":"41.871084"}}`

I have seen an issue raised on Github [#1047](https://github.com/elastic/elasticsearch-hadoop/issues/1047) that has a similar issue of latitudes that are strings being inferred as double, and while reading the data, it throws an error.

Could you please let me know how I can enforce latitude and longitude to be interpreted as Strings rather than double while creation of the DataFrame using any options.

Spark Version - 2.1.1  
ES-Hadoop jar - elasticsearch-spark-20\_2.11-5.2.2.jar  
ES Version - 5.2.2

Thanks

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 9, 2018, 2:20am UTC](https://discuss.elastic.co/t/how-to-cast-incorrectly-detected-schema-pyspark-es/123467/2 "2018-04-09T02:20:58Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.