Ошибка PyPark в отображении Spark с момента сопоставления ElasticSearch

Question

в базе данных эластичного поиска У меня есть эти данные:

{
    "titre": "Formation ElasticSearch",
    "sous-titre": "Mon sous titre",
    "formateurs": [
        {
            "prenom": "Martin",
            "nom": "Legros"
        }
    ],
    "jours": 3,
    "url": "http://test.fr"
}

Formateurs это множество людей. здесь у нас есть один человек.

и я делаю это отображение на pySpark:

person= StructType([
    StructField("nom", StringType()),
    StructField("prenom", StringType()),
])

schema= StructType([
    StructField("titre", StringType()),
    StructField("sous-titre", StringType()),
    StructField("jours", LongType()),
    StructField("url", StringType()),
    StructField("formateurs", ArrayType(person)),
])

parcel= sqlContext.read.format("org.elasticsearch.spark.sql").schema(schema).load("zenika")
parcel.printSchema()
parcel.show(1)

Я получаю эту схему:

|-- titre: string (nullable = true)
|-- sous-titre: string (nullable = true)
|-- jours: long (nullable = true)
|-- url: string (nullable = true)
|-- formateurs: array (nullable = true)
|    |-- element: struct (containsNull = true)
|    |    |-- nom: string (nullable = true)
|    |    |-- prenom: string (nullable = true)

в этом примере ошибок нет

но если я добавлю одного форматера, у меня будет одна ошибка. Пример:

{
    "titre": "Formation ElasticSearch",
    "sous-titre": "Mon sous titre",
    "formateurs": [
        {
            "prenom": "Martin",
            "nom": "Legros"
        },
        {
            "prenom": "Marc",
            "nom": "Duchien"
        }
    ],
    "jours": 3,
    "url": "http://test.fr"
}

и я получаю эту ошибку:

Caused by: org.elasticsearch.hadoop.EsHadoopIllegalStateException: Field 'formateurs.nom' not found; typically this occurs with arrays which are not mapped as single value
    at org.elasticsearch.spark.sql.RowValueReader$class.rowColumns(RowValueReader.scala:51)
    at org.elasticsearch.spark.sql.ScalaRowValueReader.rowColumns(ScalaEsRowValueReader.scala:32)
    at org.elasticsearch.spark.sql.ScalaRowValueReader.createMap(ScalaEsRowValueReader.scala:69)
    at org.elasticsearch.hadoop.serialization.ScrollReader.map(ScrollReader.java:968)
    at org.elasticsearch.hadoop.serialization.ScrollReader.readListItem(ScrollReader.java:875)
    at org.elasticsearch.hadoop.serialization.ScrollReader.list(ScrollReader.java:927)
    at org.elasticsearch.hadoop.serialization.ScrollReader.read(ScrollReader.java:833)
    at org.elasticsearch.hadoop.serialization.ScrollReader.map(ScrollReader.java:1004)
    at org.elasticsearch.hadoop.serialization.ScrollReader.read(ScrollReader.java:846)
    at org.elasticsearch.hadoop.serialization.ScrollReader.readHitAsMap(ScrollReader.java:602)
    at org.elasticsearch.hadoop.serialization.ScrollReader.readHit(ScrollReader.java:426)
    ... 27 more

Не могли бы вы объяснить мне, как создать ArrayType, потому что я не нашел учебник со сложной схемой.

Большое спасибо.

редактировать -----------------------

после исследования я нахожу это:

conf= SparkConf() \
    .set("es.read.field.as.array.include", "formateurs.nom, ...") \
    .set("es.nodes", "localhost") \
    .set( "es.port", "9200") \
    .set( "es.input.json", "yes")
sc = SparkContext(conf=conf)

просто добавьте es.read.field.as.array.include в конфигурацию SparkContext. можно добавлять вложенные объекты через запятую