Elasticsearch return empty list of objects to spark

[m-mortezapour]

I have a problem with Elasticsearch .

I want to read data from Elasticsearch index with Pyspark . my data look like as below:

user_id: 123,
features: {
    hashtags: [
        {
            text: "hello",
            count: 2
        },
        {
            text: "world",
            count: 1
        }
    ]
}
...

and when data loaded it seems Elasticsearch return empty list of objects. my dataframe after read is look like as below:

+----------+-------------------+
|  features|            user_id|
+----------+-------------------+
|{[{}, {}]}|                123|
|    {[{}]}|                384|
|    {[{}]}|                 94|
|{[{}, {}]}|                880|
+----------+-------------------+

I read data from elastic with using this configuration:

tweets = sqlContext.read.format("org.elasticsearch.spark.sql") \
    .option("es.nodes", "localhost") \
    .option("es.port", "9200") \
    .option("es.read.field.as.array.include", "features.hashtags")\
    .option("es.read.field.include", "user_id, features.hashtags")\
    .option("es.resource", "twitter")\
    .load().limit(10)

can you help me for resolve it?

[masseyke]

I'm not sure I understand the issue. Can you post exact steps to reproduce, as well as what you expect to see and what you are actually seeing?

[m-mortezapour]

thanks @masseyke for reply
I want to see a dataframe like this:

+--------------------------+-------------------+
|  features                |            user_id|
+--------------------------+-------------------+
|{[{2, hello}, {1, world}]}|                123|
|    {[{3, test1}]}        |                384|
|    {[{4, test2}]}        |                 94|
|{[{1, test3}, {1, test4}]}|                880|
+--------------------------+-------------------+

but elastic (or connector) return an empty list of object for column "features". after read data from elasticsearch my dataframe is look likes below:

+----------+-------------------+
|  features|            user_id|
+----------+-------------------+
|{[{}, {}]}|                123|
|    {[{}]}|                384|
|    {[{}]}|                 94|
|{[{}, {}]}|                880|
+----------+-------------------+

[masseyke]

@m-mortezapour can you post the mapping for your index?

[m-mortezapour]

@masseyke

"features": {
                    "properties": {
                        "hashtags": {
                            "properties": {
                                "count": {
                                    "type": "long"
                                },
                                "text": {
                                    "type": "text",
                                    "fields": {
                                        "keyword": {
                                            "type": "keyword",
                                            "ignore_above": 256
                                        }
                                    }
                                }
                            }
                        }
                    }
                },

[masseyke]

I'm not set up for python at the moment, but here's what I did:
Create the index with mapping:

curl -X PUT "localhost:9200/test?pretty" -H 'Content-Type: application/json' -d'
{
   "mappings":{
      "properties":{
         "features":{
            "properties":{
               "hashtags":{
                  "properties":{
                     "count":{
                        "type":"long"
                     },
                     "text":{
                        "type":"text",
                        "fields":{
                           "keyword":{
                              "type":"keyword",
                              "ignore_above":256
                           }
                        }
                     }
                  }
               }
            }
         }
      }
   }
}
'

Insert a document:

curl -X POST "localhost:9200/test/_doc/?pretty" -H 'Content-Type: application/json' -d'
{
  "features": {
    "hashtags": [{"text":"hello", "count":2}, {"text":"world","count":1}]
  }
}
'

Then in spark shell:

val df = sqlContext.read.format("org.elasticsearch.spark.sql").option("es.read.field.as.array.include","features.hashtags").load("test")
df.select("features.hashtags").show
+--------------------+
|            hashtags|
+--------------------+
|[{2, hello}, {1, ...|
+--------------------+

So if I'm understanding your setup correctly, the problem sounds isolated to python. I'll have to investigate that next.

[masseyke]

Maybe I'm misunderstanding the problem. It also works for me in pyspark. Could you post more details? Here's what I did in pyspark:

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.read.format("org.elasticsearch.spark.sql").option("es.read.field.as.array.include","features.hashtags").load("test")
df.select("features.hashtags").show()
+--------------------+                                                          
|            hashtags|
+--------------------+
|[{2, hello}, {1, ...|
+--------------------+

[m-mortezapour]

@masseyke Thanks so much
How can i provide you more details?

I think i found why it's happen
I create a new Index with this mapping:

{
   "mappings":{
      "properties":{
         "features":{
            "properties":{
               "hashtags":{
                  "properties":{
                     "count":{
                        "type":"long"
                     },
                     "text":{
                        "type":"text",
                        "fields":{
                           "keyword":{
                              "type":"keyword",
                              "ignore_above":256
                           }
                        }
                     }
                  }
               }
            }
         },
         "user_id": {"type": "long"},
         "another_attr": {"type": "text"}
      }
   }
}

and insert a document to index.
when i read data from pyspark (of spark shell) without option es.read.field.include it's work correctly but when add that option it return empty list.
so, when run this:

val df = sqlContext.read.format("org.elasticsearch.spark.sql").option("es.read.field.as.array.include","features.hashtags").load("new_test")

it return documents correctly:

+------------+--------------------------+-------+
|another_attr|features                  |user_id|
+------------+--------------------------+-------+
|test1       |{[{2, hello}, {1, world}]}|123    |
+------------+--------------------------+-------+

but when read data with es.read.field.include option:

val df = sqlContext.read.format("org.elasticsearch.spark.sql").option("es.read.field.include","user_id, features.hashtags").option("es.read.field.as.array.include","features.hashtags").load("new_test")

it return an empty list for features:

+----------+-------+
|features  |user_id|
+----------+-------+
|{[{}, {}]}|123    |
+----------+-------+

[masseyke]

OK I've been able to reproduce this with "es.read.field.include" set. I'll track that down in the near future. In the meantime, I assume you can work around this by not setting that?

[m-mortezapour]

@masseyke thanks a lot.
yes. without "es.read.field.include", it work.
but I need that.

[masseyke]

After digging into this a little more, I realized that the problem is an oddity with "es.read.field.include". Since you set it to "features.hashtags", it updates the schema in the dataframe to drop "count" and "text". If you print the schema it looks like this:

>>> df.printSchema()
root
 |-- features: struct (nullable = true)
 |    |-- hashtags: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)

If you print the schema without "es.read.field.include" it looks like:

>>> df.printSchema()
root
 |-- features: struct (nullable = true)
 |    |-- hashtags: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- count: long (nullable = true)
 |    |    |    |-- text: string (nullable = true)

And since those fields aren't in the schema, you don't see them when you show the dataframe. To get around this you can set "es.read.field.include" to "features.hashtags.*" so that it keeps everything in that branch of the schema. This is not exactly intuitive behavior. We need to decide what the expected behavior is here. If we leave it the same I'll try to document it better.

[masseyke]

@m-mortezapour let me know if the documentation addition at https://github.com/elastic/elasticsearch-hadoop/pull/1822/files#diff-43bc7ab0819c7ce288dc102f7cbf8ee0141fdec161c48d762db9a6232c13ec18R1581 (which will be in the 8.1.0 documentation) and the examples at #1822 (comment) are helpful. I think this is solved though, so I'm closing this ticket.