Сохранение кадра данных со столбцом массива привело к появлению org.elasticsearch.hadoop.serialization.EsHadoopSerializationException

Question

У меня есть датафрейм с массивом в столбце. Я хочу сохранить этот фрейм данных в Elasticsearch. но я получил исключение на исполнителе при сохранении кадра данных

Если столбца в виде массива нет, я могу сохранить фрейм данных. Ошибка возникает только в том случае, если какой-либо столбец находится в формате массива / вложенного / json

Вот код:

from pyspark.sql.types import *
schema = StructType([  # schema
    StructField("id", StringType(), True),
    StructField("email", ArrayType(StringType()), True)])

df = spark.createDataFrame([{"id": "id3","email": ["email1@gmail.com"]},
                            {"id": "id4", "email": ["email1@gmail.com", "email2@gmail.com"]}],
                           schema=schema)


df.show(truncate=False)


df.write.format("org.elasticsearch.spark.sql") \
                .mode('append') \
                .option("es.input.json", "yes") \
                .option("es.resource", "test/test") \
                .option("es.nodes", "10.1.129.80:9200") \
                .option("es.net.ssl", "true") \
                .option("es.net.ssl.cert.allow.self.signed", "false") \
                .option("es.net.http.auth.user", "username") \
                .option("es.net.http.auth.pass", "password") \
                .option("es.net.ssl.truststore.location", "file:///Users/alfred/Documents/DS/bin/truststore.jks") \
                .option("es.net.ssl.truststore.pass", "xxxx") \
                .option("es.net.ssl.keystore.location", "file:///Users/alfred/Documents/DS/bin/keystore.jks") \
                .option("es.net.ssl.keystore.pass", "xxxx") \
                .option("es.net.ssl.protocol","TLS") \
                .option("es.write.operation", "upsert") \
        .option("es.spark.dataframe.write.null","true")\
                .save()

Я выполнил код ниже с помощью команды ниже:

spark-submit --jars /Users/alfred/Documents/DS/bin/elasticsearch-spark-20_2.11-6.1.2.jar,/Users/alfred/Documents/DS/bin/scala-library-2.11.12.jar test_spark.py

Я получил ошибку:

19/04/27 00:48:37 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1)
org.elasticsearch.hadoop.serialization.EsHadoopSerializationException: org.codehaus.jackson.JsonParseException: Unexpected character ('(' (code 40)): expected a valid value (number, String, array, object, 'true', 'false' or 'null')
 at [Source: [B@782e5ac0; line: 1, column: 2]
    at org.elasticsearch.hadoop.serialization.json.JacksonJsonParser.nextToken(JacksonJsonParser.java:95)

Кто-нибудь может помочь решить эту проблему?